Fix regression in source IP spoofing

We were using the `cfg` in pool since istio#1040. `cfg` is not whether its
used, but rather the intent of the user -- we may enable or disable at
runtime.

This renames the field to make it clear it should not be used like this,
and fixes the issue. This went through CI due to
istio#1084.

src/config.rs

@@ -203,8 +203,9 @@ pub struct Config {
     /// Specify the number of worker threads the Tokio Runtime will use.
     pub num_worker_threads: usize,
 
-    // If true, then use original source proxying
-    pub enable_original_source: Option<bool>,
+    // If set, explicitly configure whether to use original source.
+    // If unset (recommended), this is automatically detected based on permissions.
+    pub explicitly_configure_original_source: Option<bool>,
 
     // CLI args passed to ztunnel at runtime
     pub proxy_args: String,
@@ -435,7 +436,7 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
             pc.concurrency.unwrap_or(DEFAULT_WORKER_THREADS).into(),
         )?,
 
-        enable_original_source: parse(ENABLE_ORIG_SRC)?,
+        explicitly_configure_original_source: parse(ENABLE_ORIG_SRC)?,
         proxy_args: parse_args(),
         dns_resolver_cfg,
         dns_resolver_opts,

src/proxy.rs

@@ -418,7 +418,7 @@ pub(super) fn maybe_set_transparent(
     pi: &ProxyInputs,
     listener: &socket::Listener,
 ) -> Result<bool, Error> {
-    Ok(match pi.cfg.enable_original_source {
+    Ok(match pi.cfg.explicitly_configure_original_source {
         Some(true) => {
             // Explicitly enabled. Return error if we cannot set it.
             listener.set_transparent()?;

src/proxy/inbound.rs

@@ -62,7 +62,10 @@ impl Inbound {
             .map_err(|e| Error::Bind(pi.cfg.inbound_addr, e))?;
         let transparent = super::maybe_set_transparent(&pi, &listener)?;
         // Override with our explicitly configured setting
-        let enable_orig_src = pi.cfg.enable_original_source.unwrap_or(transparent);
+        let enable_orig_src = pi
+            .cfg
+            .explicitly_configure_original_source
+            .unwrap_or(transparent);
 
         info!(
             address=%listener.local_addr(),

src/proxy/inbound_passthrough.rs

@@ -49,7 +49,10 @@ impl InboundPassthrough {
 
         let transparent = super::maybe_set_transparent(&pi, &listener)?;
         // Override with our explicitly configured setting
-        let enable_orig_src = pi.cfg.enable_original_source.unwrap_or(transparent);
+        let enable_orig_src = pi
+            .cfg
+            .explicitly_configure_original_source
+            .unwrap_or(transparent);
 
         info!(
             address=%listener.local_addr(),

src/proxy/outbound.rs

@@ -55,7 +55,10 @@ impl Outbound {
             .map_err(|e| Error::Bind(pi.cfg.outbound_addr, e))?;
         let transparent = super::maybe_set_transparent(&pi, &listener)?;
         // Override with our explicitly configured setting
-        let enable_orig_src = pi.cfg.enable_original_source.unwrap_or(transparent);
+        let enable_orig_src = pi
+            .cfg
+            .explicitly_configure_original_source
+            .unwrap_or(transparent);
 
         info!(
             address=%listener.local_addr(),
@@ -84,6 +87,7 @@ impl Outbound {
         let (sub_drain_signal, sub_drain) = drain::channel();
         let pool = proxy::pool::WorkloadHBONEPool::new(
             self.pi.cfg.clone(),
+            self.enable_orig_src,
             self.pi.socket_factory.clone(),
             self.pi.cert_manager.clone(),
         );
@@ -649,6 +653,7 @@ mod tests {
 
         let sock_fact = std::sync::Arc::new(crate::proxy::DefaultSocketFactory);
         let cert_mgr = identity::mock::new_secret_manager(Duration::from_secs(10));
+        let original_src = false; // for testing, not needed
         let outbound = OutboundConnection {
             pi: Arc::new(ProxyInputs {
                 cert_manager: identity::mock::new_secret_manager(Duration::from_secs(10)),
@@ -660,8 +665,13 @@ mod tests {
                 connection_manager: ConnectionManager::default(),
             }),
             id: TraceParent::new(),
-            pool: pool::WorkloadHBONEPool::new(cfg.clone(), sock_fact, cert_mgr.clone()),
-            enable_orig_src: cfg.enable_original_source.unwrap_or_default(),
+            pool: pool::WorkloadHBONEPool::new(
+                cfg.clone(),
+                original_src,
+                sock_fact,
+                cert_mgr.clone(),
+            ),
+            enable_orig_src: cfg.explicitly_configure_original_source.unwrap_or_default(),
             hbone_port: cfg.inbound_addr.port(),
         };
 

src/proxy/pool.rs

@@ -76,6 +76,7 @@ struct PoolState {
 
 struct ConnSpawner {
     cfg: Arc<config::Config>,
+    original_source: bool,
     socket_factory: Arc<dyn SocketFactory + Send + Sync>,
     cert_manager: Arc<SecretManager>,
     timeout_rx: watch::Receiver<bool>,
@@ -86,15 +87,13 @@ impl ConnSpawner {
     async fn new_pool_conn(&self, key: WorkloadKey) -> Result<ConnClient, Error> {
         debug!("spawning new pool conn for {}", key);
 
-        let local = self
-            .cfg
-            .enable_original_source
-            .unwrap_or_default()
-            .then_some(key.src);
+        let local = self.original_source.then_some(key.src);
         let cert = self.cert_manager.fetch_certificate(&key.src_id).await?;
         let connector = cert.outbound_connector(key.dst_id.clone())?;
         let tcp_stream =
             super::freebind_connect(local, key.dst, self.socket_factory.as_ref()).await?;
+
+        tracing::error!("freebind {:?} {}", local, tcp_stream.local_addr().unwrap());
         let tls_stream = connector.connect(tcp_stream).await?;
         trace!("connector connected, handshaking");
         let sender =
@@ -337,6 +336,7 @@ impl WorkloadHBONEPool {
     // Callers should then be safe to drop() the pool instance.
     pub fn new(
         cfg: Arc<crate::config::Config>,
+        original_source: bool,
         socket_factory: Arc<dyn SocketFactory + Send + Sync>,
         cert_manager: Arc<SecretManager>,
     ) -> WorkloadHBONEPool {
@@ -346,6 +346,7 @@ impl WorkloadHBONEPool {
 
         let spawner = ConnSpawner {
             cfg,
+            original_source,
             socket_factory,
             cert_manager,
             timeout_rx: timeout_recv.clone(),
@@ -992,7 +993,8 @@ mod test {
         };
         let sock_fact = Arc::new(crate::proxy::DefaultSocketFactory);
         let cert_mgr = identity::mock::new_secret_manager(Duration::from_secs(10));
-        let pool = WorkloadHBONEPool::new(Arc::new(cfg), sock_fact, cert_mgr);
+        let original_src = false; // for testing, not needed
+        let pool = WorkloadHBONEPool::new(Arc::new(cfg), original_src, sock_fact, cert_mgr);
         let server = TestServer {
             conn_counter,
             drop_rx,

src/proxy/socks5.rs

@@ -32,6 +32,7 @@ pub(super) struct Socks5 {
     pi: Arc<ProxyInputs>,
     listener: socket::Listener,
     drain: Watch,
+    enable_orig_src: bool,
 }
 
 impl Socks5 {
@@ -41,16 +42,23 @@ impl Socks5 {
             .tcp_bind(pi.cfg.socks5_addr.unwrap())
             .map_err(|e| Error::Bind(pi.cfg.socks5_addr.unwrap(), e))?;
 
+        let transparent = super::maybe_set_transparent(&pi, &listener)?;
+        let enable_orig_src = pi
+            .cfg
+            .explicitly_configure_original_source
+            .unwrap_or(transparent);
         info!(
             address=%listener.local_addr(),
             component="socks5",
+            transparent,
             "listener established",
         );
 
         Ok(Socks5 {
             pi,
             listener,
             drain,
+            enable_orig_src,
         })
     }
 
@@ -70,6 +78,7 @@ impl Socks5 {
                 // but ProxyInfo is overloaded and only `outbound` should ever use the pool.
                 let pool = crate::proxy::pool::WorkloadHBONEPool::new(
                     self.pi.cfg.clone(),
+                    self.enable_orig_src,
                     self.pi.socket_factory.clone(),
                     self.pi.cert_manager.clone(),
                 );
@@ -80,7 +89,7 @@ impl Socks5 {
                             pi: self.pi.clone(),
                             id: TraceParent::new(),
                             pool,
-                            enable_orig_src: self.pi.cfg.enable_original_source.unwrap_or_default(),
+                            enable_orig_src: self.enable_orig_src,
                             hbone_port: self.pi.cfg.inbound_addr.port(),
                         };
                         tokio::spawn(async move {