feat(tonic-xds): gRFC A29 data-plane TLS connector wiring

tonic-xds/src/client/channel.rs

@@ -5,9 +5,9 @@ use crate::client::lb::{ClusterDiscovery, XdsLbService};
 use crate::client::route::{Router, XdsRoutingLayer};
 use crate::xds::bootstrap::{BootstrapConfig, BootstrapError};
 use crate::xds::cache::XdsCache;
-use crate::xds::cluster_discovery::{
-    EndpointConnector, XdsClusterDiscovery, default_endpoint_connector,
-};
+#[cfg(feature = "_tls-any")]
+use crate::xds::cert_provider::{CertProviderError, CertProviderRegistry};
+use crate::xds::cluster_discovery::XdsClusterDiscovery;
 use crate::xds::resource_manager::XdsResourceManager;
 use crate::xds::routing::XdsRouter;
 use http::Request;
@@ -67,6 +67,10 @@ pub enum BuildError {
     /// Bootstrap configuration could not be loaded.
     #[error("bootstrap: {0}")]
     Bootstrap(#[from] BootstrapError),
+    /// A `certificate_providers` entry in the bootstrap failed to initialize.
+    #[cfg(feature = "_tls-any")]
+    #[error("certificate provider: {0}")]
+    CertProvider(#[from] CertProviderError),
 }
 
 /// Holds owned resources whose background tasks must live as long as the channel.
@@ -187,9 +191,10 @@ impl XdsChannelBuilder {
             )));
         }
 
-        // TODO(PR2/A29): Build CertProviderRegistry from bootstrap.certificate_providers
-        // and pass it to XdsClusterDiscovery so data-plane connections can use
-        // TLS/mTLS when CDS clusters specify UpstreamTlsContext.
+        #[cfg(feature = "_tls-any")]
+        let cert_provider_registry = Arc::new(CertProviderRegistry::from_bootstrap(
+            &bootstrap.certificate_providers,
+        )?);
 
         let node = Node::try_from(bootstrap.node)?;
         let client_config = ClientConfig::new(node, &server_uri);
@@ -200,7 +205,13 @@ impl XdsChannelBuilder {
         let resource_manager =
             XdsResourceManager::new(xds_client.clone(), cache.clone(), listener_name);
 
-        Ok(self.build_from_cache(cache, xds_client, resource_manager))
+        Ok(self.build_from_cache(
+            cache,
+            #[cfg(feature = "_tls-any")]
+            cert_provider_registry,
+            xds_client,
+            resource_manager,
+        ))
     }
 
     /// Internal builder that wires the service stack from a pre-built cache.
@@ -210,13 +221,19 @@ impl XdsChannelBuilder {
     fn build_from_cache(
         &self,
         cache: Arc<XdsCache>,
+        #[cfg(feature = "_tls-any")] cert_provider_registry: Arc<CertProviderRegistry>,
         xds_client: XdsClient,
         resource_manager: XdsResourceManager,
     ) -> XdsChannelGrpc {
         let router: Arc<dyn Router> = Arc::new(XdsRouter::new(&cache));
-        let connector: EndpointConnector = Arc::new(default_endpoint_connector);
-        let discovery: Arc<dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>> =
-            Arc::new(XdsClusterDiscovery::new(cache, connector));
+        #[cfg(feature = "_tls-any")]
+        let discovery: Arc<
+            dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
+        > = Arc::new(XdsClusterDiscovery::new(cache, cert_provider_registry));
+        #[cfg(not(feature = "_tls-any"))]
+        let discovery: Arc<
+            dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
+        > = Arc::new(XdsClusterDiscovery::new(cache));
         let retry_policy = GrpcRetryPolicy::new(GrpcRetryPolicyConfig::default());
 
         let resources = Arc::new(XdsChannelResources {
@@ -535,6 +552,19 @@ mod tests {
         assert_eq!(response.into_inner().message, "retry-server: retry-test");
     }
 
+    /// Helper: creates a minimal plaintext `ClusterResource` for tests that
+    /// drive `XdsClusterDiscovery`. The cluster watch in `discover_cluster`
+    /// blocks until a cluster is in the cache.
+    fn make_test_cluster(cluster_name: &str) -> Arc<crate::xds::resource::ClusterResource> {
+        use crate::xds::resource::cluster::{ClusterResource, LbPolicy};
+        Arc::new(ClusterResource {
+            name: cluster_name.to_string(),
+            eds_service_name: None,
+            lb_policy: LbPolicy::RoundRobin,
+            security: None,
+        })
+    }
+
     /// Helper: creates a `RouteConfigResource` that routes all traffic to the given cluster.
     fn make_test_route_config(cluster_name: &str) -> Arc<RouteConfigResource> {
         use crate::xds::resource::route_config::*;
@@ -582,15 +612,24 @@ mod tests {
     /// Builds an XdsChannelGrpc using real XdsRouter and XdsClusterDiscovery
     /// backed by the given cache.
     async fn build_xds_channel_from_cache(cache: Arc<XdsCache>) -> XdsChannelGrpc {
-        use crate::xds::cluster_discovery::{
-            EndpointConnector, XdsClusterDiscovery, default_endpoint_connector,
-        };
+        use crate::xds::cluster_discovery::XdsClusterDiscovery;
         use crate::xds::routing::XdsRouter;
 
         let router: Arc<dyn Router> = Arc::new(XdsRouter::new(&cache));
-        let connector: EndpointConnector = Arc::new(default_endpoint_connector);
-        let discovery: Arc<dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>> =
-            Arc::new(XdsClusterDiscovery::new(cache, connector));
+
+        #[cfg(feature = "_tls-any")]
+        let discovery: Arc<
+            dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
+        > = {
+            use crate::xds::cert_provider::CertProviderRegistry;
+            let registry =
+                Arc::new(CertProviderRegistry::from_bootstrap(&Default::default()).unwrap());
+            Arc::new(XdsClusterDiscovery::new(cache, registry))
+        };
+        #[cfg(not(feature = "_tls-any"))]
+        let discovery: Arc<
+            dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
+        > = Arc::new(XdsClusterDiscovery::new(cache));
 
         let builder = XdsChannelBuilder::new(test_config());
         builder.build_grpc_channel_from_parts(router, discovery, GrpcRetryPolicy::default())
@@ -608,6 +647,7 @@ mod tests {
 
         let cache = Arc::new(XdsCache::new());
         cache.update_route_config(make_test_route_config(cluster_name));
+        cache.update_cluster(cluster_name, make_test_cluster(cluster_name));
         cache.update_endpoints(cluster_name, make_test_endpoints(cluster_name, &servers));
 
         let channel = build_xds_channel_from_cache(cache).await;
@@ -641,6 +681,7 @@ mod tests {
 
         let cache = Arc::new(XdsCache::new());
         cache.update_route_config(make_test_route_config(cluster_name));
+        cache.update_cluster(cluster_name, make_test_cluster(cluster_name));
         // Start with only the first server.
         cache.update_endpoints(
             cluster_name,
@@ -692,6 +733,15 @@ mod tests {
             XdsResourceManager::new(xds_client.clone(), cache.clone(), "test-listener".into());
 
         let builder = XdsChannelBuilder::new(test_config());
+
+        #[cfg(feature = "_tls-any")]
+        let _channel = {
+            use crate::xds::cert_provider::CertProviderRegistry;
+            let registry =
+                Arc::new(CertProviderRegistry::from_bootstrap(&Default::default()).unwrap());
+            builder.build_from_cache(cache, registry, xds_client, resource_manager)
+        };
+        #[cfg(not(feature = "_tls-any"))]
         let _channel = builder.build_from_cache(cache, xds_client, resource_manager);
         // Construction should succeed without panicking.
     }

tonic-xds/src/client/endpoint.rs

@@ -100,7 +100,6 @@ pub(crate) struct EndpointChannel<S> {
 impl<S> EndpointChannel<S> {
     /// Creates a new `EndpointChannel`.
     /// This should be used by xDS implementations to construct channels to individual endpoints.
-    #[allow(dead_code)]
     pub(crate) fn new(inner: S) -> Self {
         Self {
             inner,

tonic-xds/src/lib.rs

@@ -81,6 +81,45 @@
 //! // let client = MyServiceClient::new(channel);
 //! ```
 //!
+//! ## TLS Security (gRFC A29)
+//!
+//! Upstream data-plane TLS is enabled when:
+//!
+//! 1. The crate is built with `tls-ring` *or* `tls-aws-lc` (exactly one).
+//! 2. The bootstrap JSON declares `certificate_providers` entries — each
+//!    referenced by `instance_name` in CDS resources.
+//! 3. A CDS `Cluster` carries `transport_socket: UpstreamTlsContext` naming
+//!    those instances (configured on the xDS control plane).
+//!
+//! Only the `file_watcher` plugin is built in. It reads PEM files from disk
+//! and refreshes them on `refresh_interval` (default 600s) — rotated certs
+//! reach existing TLS connectors on the next handshake.
+//!
+//! ```json
+//! {
+//!   "xds_servers": [{"server_uri": "xds.example.com:443"}],
+//!   "certificate_providers": {
+//!     "root_ca":  { "plugin_name": "file_watcher", "config": {
+//!       "ca_certificate_file": "/etc/certs/ca.pem"
+//!     }},
+//!     "identity": { "plugin_name": "file_watcher", "config": {
+//!       "certificate_file":  "/etc/certs/cert.pem",
+//!       "private_key_file":  "/etc/certs/key.pem",
+//!       "refresh_interval":  "60s"
+//!     }}
+//!   }
+//! }
+//! ```
+//!
+//! When `match_typed_subject_alt_names` is set on the cluster's validation
+//! context, the server cert's SAN list must match one of the configured
+//! matchers ("any" semantics). An empty matcher list accepts any cert
+//! chained to the configured CA roots.
+//!
+//! CDS updates that change a cluster's `transport_socket` rebuild that
+//! cluster's connector — new endpoint connections pick up the new config;
+//! existing TLS sessions continue.
+//!
 //! ## xDS features
 //!
 //! | Feature | gRFC | Status |
@@ -92,7 +131,7 @@
 //! | Weighted cluster traffic splitting | [A28] | Supported |
 //! | Case-insensitive header matching | [A63] | Supported |
 //! | Client-side P2C load balancing | | Supported |
-//! | TLS endpoint connections | [A29] | Planned |
+//! | TLS endpoint connections | [A29] | Supported |
 //! | Least-request load balancing | [A48] | Planned |
 //!
 //! [A27]: https://github.com/grpc/proposal/blob/master/A27-xds-global-load-balancing.md

tonic-xds/src/xds/bootstrap.rs

@@ -54,7 +54,9 @@ pub struct BootstrapConfig {
     ///
     /// [`CertificateProviderPluginInstance`]: https://github.com/envoyproxy/envoy/blob/main/api/envoy/extensions/transport_sockets/tls/v3/common.proto
     #[serde(default)]
-    #[allow(dead_code)] // Consumed when CertProviderRegistry is wired in (PR2/A29).
+    // Consumed by `CertProviderRegistry::from_bootstrap` only under TLS
+    // features; parsed regardless so non-TLS builds accept the same JSON.
+    #[cfg_attr(not(feature = "_tls-any"), allow(dead_code))]
     pub(crate) certificate_providers: HashMap<String, CertProviderPluginConfig>,
 }
 
@@ -105,6 +107,9 @@ pub(crate) enum ChannelCredentialType {
 ///
 /// [gRFC A29]: https://github.com/grpc/proposal/blob/master/A29-xds-tls-security.md
 #[derive(Debug, Clone, Deserialize)]
+// In non-TLS builds `cert_provider` is gated out, so nothing reads these
+// fields after serde populates them.
+#[cfg_attr(not(feature = "_tls-any"), allow(dead_code))]
 pub(crate) struct CertProviderPluginConfig {
     pub plugin_name: String,
     #[serde(default)]

tonic-xds/src/xds/cache.rs

@@ -141,7 +141,6 @@ impl XdsCache {
     }
 
     /// Watches cluster resource changes for a specific cluster.
-    #[allow(dead_code)] // Will be used when LB policy dispatch is wired (A48).
     pub(crate) fn watch_cluster(&self, name: &str) -> CacheWatch<ClusterResource> {
         self.clusters.watch(name)
     }