feat(xds): per-stream call-credentials hook for the xDS transport

tonic-xds/Cargo.toml

@@ -68,6 +68,7 @@ tonic-prost-build = "0.14"
 async-stream = "0.3"
 tempfile = "3.27.0"
 rcgen = "0.14"
+google-cloud-auth = { version = "1.9", default-features = false }
 
 [features]
 testutil = ["dep:tonic-prost"]
@@ -81,6 +82,10 @@ tls-aws-lc = ["_tls-any", "tonic/tls-aws-lc", "xds-client/tonic-tls-aws-lc"]
 name = "channel"
 required-features = ["testutil"]
 
+[[example]]
+name = "channel_with_google_default"
+required-features = ["testutil", "tls-ring"]
+
 [[example]]
 name = "greeter_server"
 required-features = ["testutil"]

tonic-xds/examples/channel_with_google_default.rs

@@ -0,0 +1,82 @@
+//! Call an xDS-fronted service through GCP Traffic Director (`google_default`).
+//!
+//! Supplies the Application Default Credentials (ADC) token as xDS call
+//! credentials by implementing `TonicCallCredentials` directly against
+//! `google-cloud-auth`.
+//!
+//! Needs a `google_default` bootstrap + ADC:
+//!
+//! ```sh
+//! GRPC_XDS_BOOTSTRAP=/path/to/bootstrap.json \
+//!     cargo run -p tonic-xds --example channel_with_google_default --features "testutil tls-ring"
+//! ```
+
+use std::sync::Arc;
+
+use google_cloud_auth::credentials::{AccessTokenCredentials, Builder};
+use tonic_xds::testutil::proto::helloworld::{HelloRequest, greeter_client::GreeterClient};
+use tonic_xds::{TonicCallCredentials, XdsChannelBuilder, XdsChannelConfig, XdsUri};
+
+const CLOUD_PLATFORM_SCOPE: &str = "https://www.googleapis.com/auth/cloud-platform";
+
+/// Fetches ADC tokens directly from `google-cloud-auth`.
+#[derive(Debug)]
+struct AdcTonicCallCredentials {
+    creds: AccessTokenCredentials,
+}
+
+impl AdcTonicCallCredentials {
+    fn new() -> std::result::Result<Self, Box<dyn std::error::Error>> {
+        let creds = Builder::default()
+            .with_scopes([CLOUD_PLATFORM_SCOPE])
+            .build_access_token_credentials()?;
+        Ok(Self { creds })
+    }
+}
+
+#[tonic::async_trait]
+impl TonicCallCredentials for AdcTonicCallCredentials {
+    async fn get_request_metadata(
+        &self,
+        metadata: &mut tonic::metadata::MetadataMap,
+    ) -> std::result::Result<(), tonic::Status> {
+        let token = self
+            .creds
+            .access_token()
+            .await
+            .map_err(|e| tonic::Status::unauthenticated(e.to_string()))?;
+        let mut value =
+            tonic::metadata::AsciiMetadataValue::try_from(format!("Bearer {}", token.token))
+                .map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
+        // Mark the bearer token sensitive so it is not accidentally logged.
+        value.set_sensitive(true);
+        metadata.insert("authorization", value);
+        Ok(())
+    }
+
+    fn requires_secure_transport(&self) -> bool {
+        true
+    }
+}
+
+#[tokio::main]
+async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
+    let target_str = std::env::var("XDS_TARGET").unwrap_or_else(|_| "xds:///my-service".into());
+    let target = XdsUri::parse(&target_str)?;
+
+    let creds: Arc<dyn TonicCallCredentials> = Arc::new(AdcTonicCallCredentials::new()?);
+
+    let channel =
+        XdsChannelBuilder::new(XdsChannelConfig::new(target).with_call_credentials(creds))
+            .build_grpc_channel()?;
+
+    let mut client = GreeterClient::new(channel);
+    let response = client
+        .say_hello(HelloRequest {
+            name: "xds-gcp".into(),
+        })
+        .await?;
+
+    println!("RESPONSE = {}", response.into_inner().message);
+    Ok(())
+}

tonic-xds/src/client/channel.rs

@@ -1,4 +1,3 @@
-use crate::XdsUri;
 use crate::client::cluster::ClusterClientRegistryGrpc;
 use crate::client::endpoint::{EndpointAddress, EndpointChannel};
 use crate::client::lb::{ClusterDiscovery, XdsLbService};
@@ -10,6 +9,7 @@ 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 crate::{TonicCallCredentials, XdsUri};
 use http::Request;
 use std::fmt::Debug;
 use std::sync::Arc;
@@ -25,6 +25,7 @@ use crate::client::retry::{GrpcRetryPolicy, GrpcRetryPolicyConfig, RetryLayer};
 pub struct XdsChannelConfig {
     target_uri: XdsUri,
     bootstrap: Option<BootstrapConfig>,
+    call_creds: Option<Arc<dyn TonicCallCredentials>>,
 }
 
 impl XdsChannelConfig {
@@ -34,6 +35,7 @@ impl XdsChannelConfig {
         Self {
             target_uri,
             bootstrap: None,
+            call_creds: None,
         }
     }
 
@@ -59,6 +61,15 @@ impl XdsChannelConfig {
         self.bootstrap = Some(BootstrapConfig::from_env()?);
         Ok(self)
     }
+
+    /// Set per-stream call credentials for the ADS stream (e.g. `google_default`).
+    ///
+    /// Attached on each (re)connect, only over a secure channel; over an insecure
+    /// channel, stream creation fails. Not refreshed mid-stream.
+    pub fn with_call_credentials(mut self, creds: Arc<dyn TonicCallCredentials>) -> Self {
+        self.call_creds = Some(creds);
+        self
+    }
 }
 
 /// Errors that can occur when building an [`XdsChannel`].
@@ -191,6 +202,10 @@ impl XdsChannelBuilder {
             )));
         }
 
+        if let Some(creds) = self.config.call_creds.clone() {
+            transport_builder = transport_builder.with_call_credentials(creds);
+        }
+
         #[cfg(feature = "_tls-any")]
         let cert_provider_registry = Arc::new(CertProviderRegistry::from_bootstrap(
             &bootstrap.certificate_providers,
@@ -723,6 +738,24 @@ mod tests {
         }
     }
 
+    #[test]
+    fn config_stores_call_credentials() {
+        #[derive(Debug)]
+        struct DummyCreds;
+        #[tonic::async_trait]
+        impl crate::TonicCallCredentials for DummyCreds {
+            async fn get_request_metadata(
+                &self,
+                _metadata: &mut tonic::metadata::MetadataMap,
+            ) -> Result<(), tonic::Status> {
+                Ok(())
+            }
+        }
+        let config = XdsChannelConfig::new(XdsUri::parse("xds:///svc").unwrap())
+            .with_call_credentials(std::sync::Arc::new(DummyCreds));
+        assert!(config.call_creds.is_some());
+    }
+
     /// Smoke test: verifies builder wiring with a disconnected XdsClient
     /// doesn't panic during construction.
     #[tokio::test]

tonic-xds/src/lib.rs

@@ -149,6 +149,7 @@ pub use client::channel::{
 };
 pub use xds::bootstrap::{BootstrapConfig, BootstrapError};
 pub use xds::uri::{XdsUri, XdsUriError};
+pub use xds_client::TonicCallCredentials;
 
 #[cfg(any(test, feature = "testutil"))]
 pub mod testutil;

tonic-xds/src/xds/bootstrap.rs

@@ -92,6 +92,7 @@ pub(crate) struct ChannelCredentialConfig {
 pub(crate) enum ChannelCredentialType {
     Insecure,
     Tls,
+    GoogleDefault,
     #[serde(untagged)]
     Unsupported(String),
 }
@@ -269,14 +270,19 @@ impl BootstrapConfig {
             .find(|t| {
                 matches!(
                     t,
-                    ChannelCredentialType::Insecure | ChannelCredentialType::Tls
+                    ChannelCredentialType::Insecure
+                        | ChannelCredentialType::Tls
+                        | ChannelCredentialType::GoogleDefault
                 )
             })
     }
 
     /// Returns `true` if the first server's selected credential is TLS.
     pub(crate) fn use_tls(&self) -> bool {
-        self.selected_credential() == Some(&ChannelCredentialType::Tls)
+        matches!(
+            self.selected_credential(),
+            Some(ChannelCredentialType::Tls | ChannelCredentialType::GoogleDefault)
+        )
     }
 }
 
@@ -363,7 +369,7 @@ mod tests {
         assert_eq!(config.xds_servers[0].channel_creds.len(), 3);
         assert!(matches!(
             config.xds_servers[0].channel_creds[0].cred_type,
-            ChannelCredentialType::Unsupported(_)
+            ChannelCredentialType::GoogleDefault
         ));
         assert_eq!(config.xds_servers[0].server_features, vec!["xds_v3"]);
         assert_eq!(config.node.id, "projects/123/nodes/456");
@@ -401,10 +407,9 @@ mod tests {
     #[test]
     fn selected_credential_first_supported_wins() {
         let config = BootstrapConfig::from_json(full_json()).unwrap();
-        // google_default skipped, tls is first supported
         assert_eq!(
             config.selected_credential(),
-            Some(&ChannelCredentialType::Tls)
+            Some(&ChannelCredentialType::GoogleDefault)
         );
     }
 
@@ -429,7 +434,7 @@ mod tests {
         let json = r#"{
             "xds_servers": [{
                 "server_uri": "localhost:5000",
-                "channel_creds": [{"type": "google_default"}]
+                "channel_creds": [{"type": "some_future_type"}]
             }],
             "node": {"id": "n1"}
         }"#;
@@ -619,6 +624,44 @@ mod tests {
         assert!(config.certificate_providers.is_empty());
     }
 
+    #[test]
+    fn parse_google_default() {
+        let json = r#"{
+            "xds_servers": [{
+                "server_uri": "xds.example.com:443",
+                "channel_creds": [{"type": "google_default"}]
+            }],
+            "node": {"id": "n1"}
+        }"#;
+        let config = BootstrapConfig::from_json(json).unwrap();
+        assert_eq!(
+            config.xds_servers[0].channel_creds[0].cred_type,
+            ChannelCredentialType::GoogleDefault
+        );
+        assert!(config.use_tls());
+        assert_eq!(
+            config.selected_credential(),
+            Some(&ChannelCredentialType::GoogleDefault)
+        );
+    }
+
+    #[test]
+    fn selected_credential_unknown_type_skipped() {
+        let json = r#"{
+            "xds_servers": [{
+                "server_uri": "localhost:5000",
+                "channel_creds": [{"type": "some_future_type"}]
+            }],
+            "node": {"id": "n1"}
+        }"#;
+        let config = BootstrapConfig::from_json(json).unwrap();
+        assert!(matches!(
+            config.xds_servers[0].channel_creds[0].cred_type,
+            ChannelCredentialType::Unsupported(_)
+        ));
+        assert_eq!(config.selected_credential(), None);
+    }
+
     #[test]
     fn multiple_certificate_provider_instances() {
         let json = r#"{

xds-client/Cargo.toml

@@ -31,6 +31,7 @@ default = ["transport-tonic", "codegen-prost"]
 transport-tonic = [
     "rt-tokio",
     "dep:tonic",
+    "tonic/codegen",
     "dep:tokio-stream",
     "dep:http",
 ]

xds-client/src/error.rs

@@ -4,6 +4,7 @@ use thiserror::Error;
 
 /// Error type for the xDS client.
 #[derive(Debug, Error)]
+#[non_exhaustive]
 pub enum Error {
     /// Failed to connect to the xDS server.
     #[error("failed to connect: {0}")]
@@ -14,6 +15,10 @@ pub enum Error {
     #[error("stream error: {0}")]
     Stream(#[from] tonic::Status),
 
+    /// Call credentials failed, or require a secure transport.
+    #[error("call credentials error: {0}")]
+    CallCredentials(String),
+
     /// The stream was closed unexpectedly.
     #[error("stream closed unexpectedly")]
     StreamClosed,

xds-client/src/lib.rs

@@ -78,7 +78,7 @@ pub use runtime::tokio::TokioRuntime;
 
 // Tonic transport
 #[cfg(feature = "transport-tonic")]
-pub use transport::tonic::{TonicTransport, TonicTransportBuilder};
+pub use transport::tonic::{TonicCallCredentials, TonicTransport, TonicTransportBuilder};
 
 // Prost codec
 #[cfg(feature = "codegen-prost")]