feat: implements ca-crl in zTunnel

Signed-off-by: nilekh <[email protected]>

Author: nilekhc

Cargo.lock

@@ -72,6 +72,8 @@ keyed_priority_queue = "0.4"
 libc = "0.2"
 log = "0.4"
 nix = { version = "0.30", features = ["socket", "sched", "uio", "fs", "ioctl", "user", "net", "mount", "resource" ] }
+notify = "6.1"
+notify-debouncer-full = "0.3"
 once_cell = "1.21"
 num_cpus = "1.16"
 ppp = "2.3"

Cargo.toml

@@ -78,6 +78,10 @@ const HTTP2_FRAME_SIZE: &str = "HTTP2_FRAME_SIZE";
 
 const UNSTABLE_ENABLE_SOCKS5: &str = "UNSTABLE_ENABLE_SOCKS5";
 
+const ENABLE_CRL: &str = "ENABLE_CRL";
+const CRL_PATH: &str = "CRL_PATH";
+const ALLOW_EXPIRED_CRL: &str = "ALLOW_EXPIRED_CRL";
+
 const DEFAULT_WORKER_THREADS: u16 = 2;
 const DEFAULT_ADMIN_PORT: u16 = 15000;
 const DEFAULT_READINESS_PORT: u16 = 15021;
@@ -108,6 +112,7 @@ const DEFAULT_ROOT_CERT_PROVIDER: &str = "./var/run/secrets/istio/root-cert.pem"
 const TOKEN_PROVIDER_ENV: &str = "AUTH_TOKEN";
 const DEFAULT_TOKEN_PROVIDER: &str = "./var/run/secrets/tokens/istio-token";
 const CERT_SYSTEM: &str = "SYSTEM";
+const DEFAULT_CRL_PATH: &str = "./var/run/secrets/istio/crl/ca-crl.pem";
 
 const PROXY_MODE_DEDICATED: &str = "dedicated";
 const PROXY_MODE_SHARED: &str = "shared";
@@ -311,6 +316,15 @@ pub struct Config {
     pub ztunnel_workload: Option<state::WorkloadInfo>,
 
     pub ipv6_enabled: bool,
+
+    /// Enable CRL (Certificate Revocation List) checking
+    pub enable_crl: bool,
+
+    /// Path to CRL file
+    pub crl_path: PathBuf,
+
+    /// Allow expired CRL (for testing/rollout scenarios)
+    pub allow_expired_crl: bool,
 }
 
 #[derive(serde::Serialize, Clone, Copy, Debug)]
@@ -865,6 +879,10 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
         ztunnel_identity,
         ztunnel_workload,
         ipv6_enabled,
+
+        enable_crl: parse_default(ENABLE_CRL, false)?,
+        crl_path: parse_default(CRL_PATH, PathBuf::from(DEFAULT_CRL_PATH))?,
+        allow_expired_crl: parse_default(ALLOW_EXPIRED_CRL, false)?,
     })
 }
 

src/config.rs

@@ -262,6 +262,10 @@ pub(super) struct ProxyInputs {
     resolver: Option<Arc<dyn Resolver + Send + Sync>>,
     // If true, inbound connections created with these inputs will not attempt to preserve the original source IP.
     pub disable_inbound_freebind: bool,
+    // CRL manager for certificate revocation checking
+    pub(super) crl_manager: Option<Arc<tls::crl::CrlManager>>,
+    // Pool registry for draining HTTP/2 connection pools on CRL reload
+    pub(super) pool_registry: pool::PoolRegistry,
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -275,6 +279,8 @@ impl ProxyInputs {
         resolver: Option<Arc<dyn Resolver + Send + Sync>>,
         local_workload_information: Arc<LocalWorkloadInformation>,
         disable_inbound_freebind: bool,
+        crl_manager: Option<Arc<tls::crl::CrlManager>>,
+        pool_registry: pool::PoolRegistry,
     ) -> Arc<Self> {
         Arc::new(Self {
             cfg,
@@ -285,6 +291,8 @@ impl ProxyInputs {
             local_workload_information,
             resolver,
             disable_inbound_freebind,
+            crl_manager,
+            pool_registry,
         })
     }
 }

src/proxy.rs

@@ -260,6 +260,39 @@ impl ConnectionManager {
         // potentially large copy under read lock, could require optimization
         self.drains.read().expect("mutex").keys().cloned().collect()
     }
+
+    /// Close all inbound and outbound connections
+    /// This is called when certificate revocations are detected
+    pub fn close_all_connections(&self) {
+        // Close all inbound connections
+        let inbound_conns = self.connections();
+        let inbound_count = inbound_conns.len();
+
+        if inbound_count > 0 {
+            tracing::info!("Closing {} inbound connection(s)", inbound_count);
+            // Spawn async tasks to close connections without blocking
+            for conn in inbound_conns {
+                let cm = self.clone();
+                tokio::spawn(async move {
+                    cm.close(&conn).await;
+                });
+            }
+        }
+
+        // Clear outbound connections (they will be recreated on next use)
+        let mut outbound = self.outbound_connections.write().expect("mutex");
+        let outbound_count = outbound.len();
+        if outbound_count > 0 {
+            tracing::info!("Clearing {} outbound connection(s)", outbound_count);
+            outbound.clear();
+        }
+
+        tracing::info!(
+            "Connection closure initiated: {} inbound, {} outbound",
+            inbound_count,
+            outbound_count
+        );
+    }
 }
 
 #[derive(serde::Serialize)]

src/proxy/connection_manager.rs

@@ -83,6 +83,7 @@ impl Inbound {
         let pi = self.pi.clone();
         let acceptor = InboundCertProvider {
             local_workload: self.pi.local_workload_information.clone(),
+            crl_manager: self.pi.crl_manager.clone(),
         };
 
         // Safety: we set nodelay directly in tls_server, so it is safe to convert to a normal listener.
@@ -683,6 +684,7 @@ impl InboundFlagError {
 #[derive(Clone)]
 struct InboundCertProvider {
     local_workload: Arc<LocalWorkloadInformation>,
+    crl_manager: Option<Arc<tls::crl::CrlManager>>,
 }
 
 #[async_trait::async_trait]
@@ -693,7 +695,7 @@ impl crate::tls::ServerCertProvider for InboundCertProvider {
             "fetching cert"
         );
         let cert = self.local_workload.fetch_certificate().await?;
-        Ok(Arc::new(cert.server_config()?))
+        Ok(Arc::new(cert.server_config(self.crl_manager.clone())?))
     }
 }
 
@@ -914,6 +916,8 @@ mod tests {
             None,
             local_workload,
             false,
+            None,
+            crate::proxy::pool::PoolRegistry::new(),
         ));
         let inbound_request = Inbound::build_inbound_request(&pi, conn, &request_parts).await;
         match want {

src/proxy/inbound.rs

@@ -79,7 +79,12 @@ impl Outbound {
             self.pi.cfg.clone(),
             self.pi.socket_factory.clone(),
             self.pi.local_workload_information.clone(),
+            self.pi.crl_manager.clone(),
         );
+
+        // Register pool with registry for CRL-triggered draining
+        self.pi.pool_registry.register(pool.clone());
+
         let pi = self.pi.clone();
         let accept = async move |drain: DrainWatcher, force_shutdown: watch::Receiver<()>| {
             loop {
@@ -247,7 +252,8 @@ impl OutboundConnection {
             .local_workload_information
             .fetch_certificate()
             .await?;
-        let connector = cert.outbound_connector(wl_key.dst_id.clone())?;
+        let connector =
+            cert.outbound_connector(wl_key.dst_id.clone(), self.pi.crl_manager.clone())?;
         let tls_stream = connector.connect(upgraded).await?;
 
         // Spawn inner CONNECT tunnel
@@ -809,6 +815,7 @@ mod tests {
                 cfg.clone(),
                 sock_fact,
                 local_workload_information.clone(),
+                None,
             ),
             hbone_port: cfg.inbound_addr.port(),
         };