chore(rtx-xdp): move init as early as possible in process start

t-nelson · t-nelson · commit 3ed065ae4604 · 2026-01-21T16:22:20.000-07:00
diff --git a/core/src/validator.rs b/core/src/validator.rs
@@ -141,7 +141,7 @@ use {
     solana_turbine::{
         self,
         broadcast_stage::BroadcastStageType,
-        xdp::{master_ip_if_bonded, XdpConfig, XdpRetransmitBuilder, XdpRetransmitter},
+        xdp::{XdpRetransmitBuilder, XdpRetransmitter},
     },
     solana_unified_scheduler_logic::SchedulingMode,
     solana_unified_scheduler_pool::{DefaultSchedulerPool, SupportedSchedulingMode},
@@ -151,7 +151,7 @@ use {
     std::{
         borrow::Cow,
         collections::{HashMap, HashSet},
-        net::{IpAddr, SocketAddr},
+        net::SocketAddr,
         num::{NonZeroU64, NonZeroUsize},
         path::{Path, PathBuf},
         str::FromStr,
@@ -412,7 +412,6 @@ pub struct ValidatorConfig {
     pub replay_transactions_threads: NonZeroUsize,
     pub tvu_shred_sigverify_threads: NonZeroUsize,
     pub delay_leader_block_for_pending_fork: bool,
-    pub retransmit_xdp: Option<XdpConfig>,
     pub repair_handler_type: RepairHandlerType,
 }
 
@@ -494,7 +493,6 @@ impl ValidatorConfig {
             tvu_shred_sigverify_threads: NonZeroUsize::new(get_thread_count())
                 .expect("thread count is non-zero"),
             delay_leader_block_for_pending_fork: false,
-            retransmit_xdp: None,
             repair_handler_type: RepairHandlerType::default(),
         }
     }
@@ -701,6 +699,7 @@ impl Validator {
         socket_addr_space: SocketAddrSpace,
         tpu_config: ValidatorTpuConfig,
         admin_rpc_service_post_init: Arc<RwLock<Option<AdminRpcRequestMetadataPostInit>>>,
+        maybe_xdp_retransmit_builder: Option<XdpRetransmitBuilder>,
     ) -> Result<Self> {
         #[cfg(debug_assertions)]
         const DEBUG_ASSERTION_STATUS: &str = "enabled";
@@ -1543,28 +1542,13 @@ impl Validator {
             )
         });
 
-        let (xdp_retransmitter, xdp_sender) = if let Some(xdp_config) =
-            config.retransmit_xdp.clone()
-        {
-            let src_port = node.sockets.retransmit_sockets[0]
-                .local_addr()
-                .expect("failed to get local address")
-                .port();
-            let src_ip = match node.bind_ip_addrs.active() {
-                IpAddr::V4(ip) if !ip.is_unspecified() => Some(ip),
-                IpAddr::V4(_unspecified) => xdp_config
-                    .interface
-                    .as_ref()
-                    .and_then(|iface| master_ip_if_bonded(iface)),
-                _ => panic!("IPv6 not supported"),
+        let (xdp_retransmitter, xdp_sender) =
+            if let Some(xdp_retransmit_builder) = maybe_xdp_retransmit_builder {
+                let (rtx, sender) = xdp_retransmit_builder.build(exit.clone());
+                (Some(rtx), Some(sender))
+            } else {
+                (None, None)
             };
-            let xdp_retransmit_builder = XdpRetransmitBuilder::new(xdp_config, src_port, src_ip)
-                .expect("failed to create xdp retransmitter");
-            let (rtx, sender) = xdp_retransmit_builder.build(exit.clone());
-            (Some(rtx), Some(sender))
-        } else {
-            (None, None)
-        };
 
         // disable all2all tests if not allowed for a given cluster type
         let alpenglow_socket = if genesis_config.cluster_type == ClusterType::Testnet
@@ -2968,6 +2952,7 @@ mod tests {
             SocketAddrSpace::Unspecified,
             ValidatorTpuConfig::new_for_tests(),
             Arc::new(RwLock::new(None)),
+            None,
         )
         .expect("assume successful validator start");
         assert_eq!(
@@ -3182,6 +3167,7 @@ mod tests {
                     SocketAddrSpace::Unspecified,
                     ValidatorTpuConfig::new_for_tests(),
                     Arc::new(RwLock::new(None)),
+                    None,
                 )
                 .expect("assume successful validator start")
             })
diff --git a/local-cluster/src/local_cluster.rs b/local-cluster/src/local_cluster.rs
@@ -425,6 +425,7 @@ impl LocalCluster {
                 // to use the same QUIC ports due to SO_REUSEPORT.
                 ValidatorTpuConfig::new_for_tests(),
                 Arc::new(RwLock::new(None)),
+                None,
             )
             .expect("assume successful validator start");
 
@@ -636,6 +637,7 @@ impl LocalCluster {
             socket_addr_space,
             ValidatorTpuConfig::new_for_tests(),
             Arc::new(RwLock::new(None)),
+            None,
         )
         .expect("assume successful validator start");
 
@@ -702,6 +704,7 @@ impl LocalCluster {
                 socket_addr_space,
                 ValidatorTpuConfig::new_for_tests(),
                 Arc::new(RwLock::new(None)),
+                None,
             )
             .unwrap_or_else(|e| panic!("Cluster leader failed to start: {e:?}"));
 
@@ -768,6 +771,7 @@ impl LocalCluster {
                     socket_addr_space,
                     ValidatorTpuConfig::new_for_tests(),
                     Arc::new(RwLock::new(None)),
+                    None,
                 )
                 .unwrap_or_else(|e| panic!("Validator {i} failed to start: {e:?}"));
 
@@ -1321,6 +1325,7 @@ impl Cluster for LocalCluster {
             socket_addr_space,
             ValidatorTpuConfig::new_for_tests(),
             Arc::new(RwLock::new(None)),
+            None,
         )
         .expect("assume successful validator start");
         cluster_validator_info.validator = Some(restarted_node);
diff --git a/local-cluster/src/validator_configs.rs b/local-cluster/src/validator_configs.rs
@@ -79,7 +79,6 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
         replay_transactions_threads: config.replay_transactions_threads,
         tvu_shred_sigverify_threads: config.tvu_shred_sigverify_threads,
         delay_leader_block_for_pending_fork: config.delay_leader_block_for_pending_fork,
-        retransmit_xdp: config.retransmit_xdp.clone(),
         repair_handler_type: config.repair_handler_type.clone(),
     }
 }
diff --git a/test-validator/src/lib.rs b/test-validator/src/lib.rs
@@ -1227,6 +1227,7 @@ impl TestValidator {
             socket_addr_space,
             ValidatorTpuConfig::new_for_tests(),
             config.admin_rpc_service_post_init.clone(),
+            None,
         )?);
 
         let test_validator = TestValidator {
diff --git a/validator/src/admin_rpc_service.rs b/validator/src/admin_rpc_service.rs
@@ -1618,6 +1618,7 @@ mod tests {
                 SocketAddrSpace::Unspecified,
                 ValidatorTpuConfig::new_for_tests(),
                 post_init.clone(),
+                None,
             )
             .expect("assume successful validator start");
             assert_eq!(
diff --git a/validator/src/commands/run/execute.rs b/validator/src/commands/run/execute.rs
@@ -66,7 +66,7 @@ use {
     },
     solana_turbine::{
         broadcast_stage::BroadcastStageType,
-        xdp::{set_cpu_affinity, XdpConfig},
+        xdp::{master_ip_if_bonded, set_cpu_affinity, XdpConfig, XdpRetransmitBuilder},
     },
     solana_validator_exit::Exit,
     std::{
@@ -269,6 +269,42 @@ pub fn execute(
 
     let mut node = Node::new_with_external_ip(&identity_keypair.pubkey(), node_config);
 
+    // XDP _MUST_ be setup _BEFORE_ the app spawns any threads to ensure linux
+    // capabilities do not leak, leaving the process in a state where it could
+    // potentially be used as a privilege escalation gadget
+    let maybe_xdp_retransmit_builder = retransmit_xdp.clone().map(|xdp_config| {
+        let src_port = node.sockets.retransmit_sockets[0]
+            .local_addr()
+            .expect("failed to get local address")
+            .port();
+        let src_ip = match node.bind_ip_addrs.active() {
+            IpAddr::V4(ip) if !ip.is_unspecified() => Some(ip),
+            IpAddr::V4(_unspecified) => xdp_config
+                .interface
+                .as_ref()
+                .and_then(|iface| master_ip_if_bonded(iface)),
+            _ => panic!("IPv6 not supported"),
+        };
+        XdpRetransmitBuilder::new(xdp_config, src_port, src_ip)
+            .expect("failed to create xdp retransmitter")
+    });
+
+    let reserved = retransmit_xdp
+        .map(|xdp| xdp.cpus.clone())
+        .unwrap_or_default()
+        .iter()
+        .cloned()
+        .collect::<HashSet<_>>();
+    if !reserved.is_empty() {
+        let available = core_affinity::get_core_ids()
+            .unwrap_or_default()
+            .into_iter()
+            .map(|core_id| core_id.id)
+            .collect::<HashSet<_>>();
+        let available = available.difference(&reserved);
+        set_cpu_affinity(available.into_iter().copied()).unwrap();
+    }
+
     solana_core::validator::report_target_features();
 
     let authorized_voter_keypairs = keypairs_of(matches, "authorized_voter_keypairs")
@@ -714,7 +750,6 @@ pub fn execute(
         wen_restart_proto_path: value_t!(matches, "wen_restart", PathBuf).ok(),
         wen_restart_coordinator: value_t!(matches, "wen_restart_coordinator", Pubkey).ok(),
         turbine_disabled: Arc::<AtomicBool>::default(),
-        retransmit_xdp,
         broadcast_stage_type: BroadcastStageType::Standard,
         block_verification_method: value_t_or_exit!(
             matches,
@@ -751,24 +786,6 @@ pub fn execute(
         .into(),
     };
 
-    let reserved = validator_config
-        .retransmit_xdp
-        .as_ref()
-        .map(|xdp| xdp.cpus.clone())
-        .unwrap_or_default()
-        .iter()
-        .cloned()
-        .collect::<HashSet<_>>();
-    if !reserved.is_empty() {
-        let available = core_affinity::get_core_ids()
-            .unwrap_or_default()
-            .into_iter()
-            .map(|core_id| core_id.id)
-            .collect::<HashSet<_>>();
-        let available = available.difference(&reserved);
-        set_cpu_affinity(available.into_iter().copied()).unwrap();
-    }
-
     let vote_account = pubkey_of(matches, "vote_account").unwrap_or_else(|| {
         if !validator_config.voting_disabled {
             warn!("--vote-account not specified, validator will not vote");
@@ -1031,6 +1048,7 @@ pub fn execute(
             vote_quic_server_config,
         },
         admin_service_post_init,
+        maybe_xdp_retransmit_builder,
     ) {
         Ok(validator) => Ok(validator),
         Err(err) => {

Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,6 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {`
`79`	`79`	`replay_transactions_threads: config.replay_transactions_threads,`
`80`	`80`	`tvu_shred_sigverify_threads: config.tvu_shred_sigverify_threads,`
`81`	`81`	`delay_leader_block_for_pending_fork: config.delay_leader_block_for_pending_fork,`
`82`		`- retransmit_xdp: config.retransmit_xdp.clone(),`
`83`	`82`	`repair_handler_type: config.repair_handler_type.clone(),`
`84`	`83`	`}`
`85`	`84`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1618,6 +1618,7 @@ mod tests {`
`1618`	`1618`	`SocketAddrSpace::Unspecified,`
`1619`	`1619`	`ValidatorTpuConfig::new_for_tests(),`
`1620`	`1620`	`post_init.clone(),`
	`1621`	`+ None,`
`1621`	`1622`	`)`
`1622`	`1623`	`.expect("assume successful validator start");`
`1623`	`1624`	`assert_eq!(`