[MFP] allow list for tx submission (#23853)

## Description 

Set an allow list to submit transactions via transaction driver to
specific validators only. The provided list in the configuration is
using the validator display names, as this is quite straightforward for
someone to setup instead of dealing with serialised keys/addresses etc.
The solution does not update the list when the committee updates, but
this could be easily done if required. For now opted in for a simpler
solution.

## Test plan 

CI/PT

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

Author: akichidis

crates/sui-config/src/node.rs

@@ -221,6 +221,19 @@ pub struct NodeConfig {
     /// Fork recovery configuration for handling validator equivocation after forks
     #[serde(skip_serializing_if = "Option::is_none")]
     pub fork_recovery: Option<ForkRecoveryConfig>,
+
+    /// Configuration for the transaction driver.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub transaction_driver_config: Option<TransactionDriverConfig>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct TransactionDriverConfig {
+    /// The list of validators that are allowed to submit MFP transactions to (via the transaction driver).
+    /// Each entry is a validator display name.
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub allowed_submission_validators: Vec<String>,
 }
 
 #[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq)]

crates/sui-core/src/transaction_driver/mod.rs

@@ -24,7 +24,6 @@ use mysten_metrics::{monitored_future, spawn_logged_monitored_task};
 use parking_lot::Mutex;
 use rand::Rng;
 use sui_types::{
-    base_types::AuthorityName,
     committee::EpochId,
     error::{ErrorCategory, UserInputError},
     messages_grpc::{PingType, SubmitTxRequest, SubmitTxResult, TxType},
@@ -55,7 +54,7 @@ pub struct SubmitTransactionOptions {
 
     /// When submitting a transaction, only the validators in the allowed validator list can be used to submit the transaction to.
     /// When the allowed validator list is empty, any validator can be used.
-    pub allowed_validators: Vec<AuthorityName>,
+    pub allowed_validators: Vec<String>,
 }
 
 #[derive(Clone, Debug)]
@@ -189,7 +188,7 @@ where
                     .drive_transaction(
                         SubmitTxRequest::new_ping(ping_type),
                         SubmitTransactionOptions {
-                            allowed_validators: vec![name],
+                            allowed_validators: vec![display_name.clone()],
                             ..Default::default()
                         },
                         Some(ping_timeout),

crates/sui-core/src/transaction_driver/request_retrier.rs

@@ -41,7 +41,7 @@ impl<A: Clone> RequestRetrier<A> {
         auth_agg: &Arc<AuthorityAggregator<A>>,
         client_monitor: &Arc<ValidatorClientMonitor<A>>,
         tx_type: TxType,
-        allowed_validators: Vec<AuthorityName>,
+        allowed_validators: Vec<String>,
     ) -> Self {
         let ranked_validators = client_monitor.select_shuffled_preferred_validators(
             &auth_agg.committee,
@@ -50,7 +50,10 @@ impl<A: Clone> RequestRetrier<A> {
         );
         let ranked_clients = ranked_validators
             .into_iter()
-            .filter(|name| allowed_validators.is_empty() || allowed_validators.contains(name))
+            .filter(|name| {
+                let display_name = auth_agg.get_display_name(name);
+                allowed_validators.is_empty() || allowed_validators.contains(&display_name)
+            })
             .filter_map(|name| {
                 // There is not guarantee that the `name` are in the `auth_agg.authority_clients` if those are coming from the list
                 // of `allowed_validators`, as the provided `auth_agg` might have been updated with a new committee that doesn't contain the validator in question.
@@ -139,7 +142,10 @@ impl<A: Clone> RequestRetrier<A> {
 
 #[cfg(test)]
 mod tests {
-    use sui_types::error::{SuiError, UserInputError};
+    use sui_types::{
+        base_types::ConciseableName,
+        error::{SuiError, UserInputError},
+    };
 
     use crate::{
         authority_aggregator::{AuthorityAggregatorBuilder, TimeoutConfig},
@@ -199,9 +205,9 @@ mod tests {
         println!("Case 1. Mix of unknown validators and one known validator");
         {
             let allowed_validators = vec![
-                unknown_validator1,
-                unknown_validator2,
-                authorities[0], // This one exists in auth_agg
+                unknown_validator1.concise().to_string(),
+                unknown_validator2.concise().to_string(),
+                authorities[0].concise().to_string(), // This one exists in auth_agg
             ];
 
             let retrier = RequestRetrier::new(
@@ -218,7 +224,10 @@ mod tests {
 
         println!("Case 2. Only unknown validators are provided");
         {
-            let allowed_validators = vec![unknown_validator1, unknown_validator2];
+            let allowed_validators = vec![
+                unknown_validator1.concise().to_string(),
+                unknown_validator2.concise().to_string(),
+            ];
 
             let retrier = RequestRetrier::new(
                 &auth_agg,

crates/sui-core/src/transaction_orchestrator.rs

@@ -78,6 +78,7 @@ pub struct TransactionOrchestrator<A: Clone> {
     metrics: Arc<TransactionOrchestratorMetrics>,
     transaction_driver: Option<Arc<TransactionDriver<A>>>,
     td_percentage: u8,
+    td_allowed_submission_list: Vec<String>,
 }
 
 impl TransactionOrchestrator<NetworkAuthorityClient> {
@@ -166,6 +167,12 @@ where
             None
         };
 
+        let td_allowed_submission_list = node_config
+            .transaction_driver_config
+            .as_ref()
+            .map(|config| config.allowed_submission_validators.clone())
+            .unwrap_or_default();
+
         Self {
             quorum_driver_handler,
             validator_state,
@@ -175,6 +182,7 @@ where
             metrics,
             transaction_driver,
             td_percentage,
+            td_allowed_submission_list,
         }
     }
 }
@@ -654,8 +662,7 @@ where
                 SubmitTxRequest::new_transaction(request.transaction.clone()),
                 SubmitTransactionOptions {
                     forwarded_client_addr: client_addr,
-                    // TODO: provide a list of validators to submit the transaction via config
-                    allowed_validators: vec![],
+                    allowed_validators: self.td_allowed_submission_list.clone(),
                 },
                 timeout_duration,
             )

crates/sui-swarm-config/src/node_config_builder.rs

@@ -14,7 +14,7 @@ use sui_config::node::{
     ExecutionTimeObserverConfig, ExpensiveSafetyCheckConfig, Genesis, KeyPairWithPath,
     StateSnapshotConfig, DEFAULT_GRPC_CONCURRENCY_LIMIT,
 };
-use sui_config::node::{default_zklogin_oauth_providers, RunWithRange};
+use sui_config::node::{default_zklogin_oauth_providers, RunWithRange, TransactionDriverConfig};
 use sui_config::p2p::{P2pConfig, SeedPeer, StateSyncConfig};
 use sui_config::verifier_signing_config::VerifierSigningConfig;
 use sui_config::{
@@ -255,6 +255,7 @@ impl ValidatorConfigBuilder {
             chain_override_for_testing: self.chain_override,
             validator_client_monitor_config: None,
             fork_recovery: None,
+            transaction_driver_config: None,
         }
     }
 
@@ -292,6 +293,7 @@ pub struct FullnodeConfigBuilder {
     data_ingestion_dir: Option<PathBuf>,
     disable_pruning: bool,
     chain_override: Option<Chain>,
+    transaction_driver_config: Option<TransactionDriverConfig>,
 }
 
 impl FullnodeConfigBuilder {
@@ -415,6 +417,14 @@ impl FullnodeConfigBuilder {
         self
     }
 
+    pub fn with_transaction_driver_config(
+        mut self,
+        config: Option<TransactionDriverConfig>,
+    ) -> Self {
+        self.transaction_driver_config = config;
+        self
+    }
+
     pub fn build<R: rand::RngCore + rand::CryptoRng>(
         self,
         rng: &mut R,
@@ -564,6 +574,7 @@ impl FullnodeConfigBuilder {
             chain_override_for_testing: self.chain_override,
             validator_client_monitor_config: None,
             fork_recovery: None,
+            transaction_driver_config: self.transaction_driver_config,
         }
     }
 }