diff --git a/tonic-xds/Cargo.toml b/tonic-xds/Cargo.toml index 3559db617..cf1c6fe0a 100644 --- a/tonic-xds/Cargo.toml +++ b/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"] diff --git a/tonic-xds/examples/channel_with_google_default.rs b/tonic-xds/examples/channel_with_google_default.rs new file mode 100644 index 000000000..de6bfc5e3 --- /dev/null +++ b/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> { + 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> { + let target_str = std::env::var("XDS_TARGET").unwrap_or_else(|_| "xds:///my-service".into()); + let target = XdsUri::parse(&target_str)?; + + let creds: Arc = 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(()) +} diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 5132d128a..49e0387cd 100644 --- a/tonic-xds/src/client/channel.rs +++ b/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, + call_creds: Option>, } 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) -> 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] diff --git a/tonic-xds/src/lib.rs b/tonic-xds/src/lib.rs index d5be6be53..4fd4ca5d2 100644 --- a/tonic-xds/src/lib.rs +++ b/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; diff --git a/tonic-xds/src/xds/bootstrap.rs b/tonic-xds/src/xds/bootstrap.rs index 913e2a7bb..65a6871b2 100644 --- a/tonic-xds/src/xds/bootstrap.rs +++ b/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#"{ diff --git a/xds-client/Cargo.toml b/xds-client/Cargo.toml index 14bb409fe..f8cc9e3d8 100644 --- a/xds-client/Cargo.toml +++ b/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", ] diff --git a/xds-client/src/error.rs b/xds-client/src/error.rs index 233dfb577..0731a8216 100644 --- a/xds-client/src/error.rs +++ b/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, diff --git a/xds-client/src/lib.rs b/xds-client/src/lib.rs index 5348c9b4d..7b290133b 100644 --- a/xds-client/src/lib.rs +++ b/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")] diff --git a/xds-client/src/transport/tonic.rs b/xds-client/src/transport/tonic.rs index 06a9f9319..f10214665 100644 --- a/xds-client/src/transport/tonic.rs +++ b/xds-client/src/transport/tonic.rs @@ -9,6 +9,7 @@ use crate::error::{Error, Result}; use crate::transport::{Transport, TransportBuilder, TransportStream}; use bytes::{Buf, BufMut, Bytes}; use http::uri::PathAndQuery; +use std::sync::Arc; use tokio::sync::mpsc; use tokio_stream::StreamExt as _; use tonic::client::Grpc; @@ -16,6 +17,25 @@ use tonic::codec::{Codec, DecodeBuf, Decoder, EncodeBuf, Encoder}; use tonic::transport::{Channel, Endpoint}; use tonic::{Status, Streaming}; +/// Per-stream call credentials for the ADS stream (e.g. a bearer token). +/// +/// Attached on each (re)connect, only when the channel is secure. +#[tonic::async_trait] +pub trait TonicCallCredentials: Send + Sync + std::fmt::Debug + 'static { + /// Generates the authentication metadata for a specific call. + async fn get_request_metadata( + &self, + metadata: &mut tonic::metadata::MetadataMap, + ) -> std::result::Result<(), Status>; + + /// Whether these credentials require a secure (TLS) transport. + fn requires_secure_transport(&self) -> bool { + // Note: a bool simplification of the `grpc` crate's + // `CallCredentials::minimum_channel_security_level` (`SecurityLevel`). + true + } +} + /// The gRPC path for the ADS StreamAggregatedResources RPC. const ADS_PATH: &str = "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"; @@ -80,36 +100,15 @@ impl Decoder for BytesDecoder { #[derive(Clone, Debug)] pub struct TonicTransport { channel: Channel, + secure: bool, + call_creds: Option>, } impl TonicTransport { - /// Create a transport from an existing tonic [`Channel`]. - /// - /// Use this when you need custom channel configuration (e.g., TLS, timeouts). - /// - /// # Example - /// - /// ```ignore - /// use tonic::transport::{Certificate, Channel, ClientTlsConfig}; - /// - /// let tls = ClientTlsConfig::new() - /// .ca_certificate(Certificate::from_pem(ca_cert)) - /// .domain_name("xds.example.com"); - /// - /// let channel = Channel::from_static("https://xds.example.com:443") - /// .tls_config(tls)? - /// .connect() - /// .await?; - /// - /// let transport = TonicTransport::from_channel(channel); - /// ``` - pub fn from_channel(channel: Channel) -> Self { - Self { channel } - } - /// Connect to an xDS server with default settings. /// - /// For custom configuration (TLS, timeouts, etc.), use [`from_channel`](Self::from_channel). + /// For custom configuration (TLS, a pre-built channel, call credentials), use + /// [`TonicTransportBuilder`]. pub async fn connect(uri: impl Into) -> Result { let server = ServerConfig::new(uri.into()); TonicTransportBuilder::new().build(&server).await @@ -151,6 +150,14 @@ pub struct TonicTransportBuilder { // - Per-server credential overrides (via ServerConfig.extensions) #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))] tls_config: Option, + + /// A pre-built channel and whether it is secure (TLS). When set, [`build`] + /// uses it as-is and ignores the server URI. + /// + /// Note: a minimal hook; the `grpc` crate provides the richer pre-built-channel pattern. + channel: Option<(Channel, bool)>, + /// Per-stream call credentials for the ADS stream. + call_creds: Option>, } impl TonicTransportBuilder { @@ -168,15 +175,79 @@ impl TonicTransportBuilder { self.tls_config = Some(tls_config); 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) -> Self { + self.call_creds = Some(creds); + self + } + + /// Use a pre-built tonic [`Channel`] instead of connecting to the server URI. + /// + /// Use this for custom channel configuration (e.g. TLS, timeouts). `secure` + /// declares whether the channel uses TLS; call credentials that require a + /// secure transport are refused when it is `false`. + /// + /// # Example + /// + /// ```ignore + /// use tonic::transport::{Channel, ClientTlsConfig}; + /// use xds_client::TonicTransportBuilder; + /// + /// let channel = Channel::from_static("https://xds.example.com:443") + /// .tls_config(ClientTlsConfig::new())? + /// .connect() + /// .await?; + /// + /// let builder = TonicTransportBuilder::new().with_channel(channel, true); + /// ``` + pub fn with_channel(mut self, channel: Channel, secure: bool) -> Self { + self.channel = Some((channel, secure)); + self + } + + /// Prepend `https://` to a scheme-less `server_uri` on the secure path. + /// + /// Bootstrap URIs like `trafficdirector.googleapis.com:443` parse with no scheme, + /// so `Endpoint` won't negotiate TLS. A scheme lets it, and tonic derive SNI from + /// `uri.host()`. Non-`http::Uri` inputs (`unix://`) and plaintext are left as-is. + fn ensure_secure_server_uri(raw: &str, secure: bool) -> String { + if secure + && let Ok(uri) = raw.parse::() + && uri.scheme().is_none() + { + return format!("https://{raw}"); + } + raw.to_string() + } } impl TransportBuilder for TonicTransportBuilder { type Transport = TonicTransport; async fn build(&self, server: &ServerConfig) -> Result { + // Use a pre-built channel as-is; the server URI is ignored. + if let Some((channel, secure)) = &self.channel { + return Ok(TonicTransport { + channel: channel.clone(), + secure: *secure, + call_creds: self.call_creds.clone(), + }); + } + + // The channel is secure only when TLS is configured; with no TLS backend + // compiled in, it can never be secure. + #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))] + let secure = self.tls_config.is_some(); + #[cfg(not(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc")))] + let secure = false; + // `Endpoint::from_shared` routes `unix://` URIs to tonic's UDS connector. // Required for control planes like Istio's grpc-agent that ship `unix:///etc/istio/proxy/XDS`. - let endpoint = Endpoint::from_shared(server.uri().to_string()) + let endpoint = Endpoint::from_shared(Self::ensure_secure_server_uri(server.uri(), secure)) .map_err(|e| Error::Connection(e.to_string()))?; #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))] @@ -192,7 +263,11 @@ impl TransportBuilder for TonicTransportBuilder { .await .map_err(|e| Error::Connection(e.to_string()))?; - Ok(TonicTransport::from_channel(channel)) + Ok(TonicTransport { + channel, + call_creds: self.call_creds.clone(), + secure, + }) } } @@ -200,6 +275,16 @@ impl Transport for TonicTransport { type Stream = TonicAdsStream; async fn new_stream(&self, initial_requests: Vec) -> Result { + // Guard before connecting: never send credentials over an insecure channel. + if let Some(creds) = &self.call_creds + && creds.requires_secure_transport() + && !self.secure + { + return Err(Error::CallCredentials( + "call credentials require a secure transport".into(), + )); + } + let mut grpc = Grpc::new(self.channel.clone()); grpc.ready() @@ -217,9 +302,18 @@ impl Transport for TonicTransport { let request_stream = initial_stream.chain(channel_stream); let path = PathAndQuery::from_static(ADS_PATH); + let mut request = tonic::Request::new(request_stream); + + // Inject the configured call credentials. + if let Some(creds) = &self.call_creds { + creds + .get_request_metadata(request.metadata_mut()) + .await + .map_err(|e| Error::CallCredentials(e.to_string()))?; + } let response = grpc - .streaming(tonic::Request::new(request_stream), path, BytesCodec) + .streaming(request, path, BytesCodec) .await .map_err(Error::Stream)?; @@ -266,13 +360,17 @@ mod tests { use prost::Message; use std::net::SocketAddr; use std::pin::Pin; + use std::sync::Arc; use tokio::net::TcpListener; use tokio_stream::Stream; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Request, Response, Status}; /// Mock ADS server that echoes back a response for each request. - struct MockAdsServer; + #[derive(Default)] + struct MockAdsServer { + expected_auth: Option, + } #[tonic::async_trait] impl AggregatedDiscoveryService for MockAdsServer { @@ -283,6 +381,17 @@ mod tests { &self, request: Request>, ) -> std::result::Result, Status> { + if let Some(expected) = &self.expected_auth { + let got = request + .metadata() + .get("authorization") + .and_then(|v| v.to_str().ok()); + if got != Some(expected.as_str()) { + return Err(Status::unauthenticated( + "missing or unexpected authorization", + )); + } + } let mut inbound = request.into_inner(); let outbound = async_stream::try_stream! { @@ -313,13 +422,16 @@ mod tests { } } - async fn start_mock_server() -> SocketAddr { + async fn start_mock_server(expected_auth: Option<&str>) -> SocketAddr { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); + let server = MockAdsServer { + expected_auth: expected_auth.map(str::to_owned), + }; tokio::spawn(async move { tonic::transport::Server::builder() - .add_service(AggregatedDiscoveryServiceServer::new(MockAdsServer)) + .add_service(AggregatedDiscoveryServiceServer::new(server)) .serve_with_incoming(TcpListenerStream::new(listener)) .await .unwrap(); @@ -330,9 +442,124 @@ mod tests { addr } + #[derive(Debug)] + struct MockCreds { + pairs: Vec<(String, String)>, + requires_secure: bool, + } + + #[tonic::async_trait] + impl TonicCallCredentials for MockCreds { + async fn get_request_metadata( + &self, + metadata: &mut tonic::metadata::MetadataMap, + ) -> std::result::Result<(), tonic::Status> { + for (name, value) in &self.pairs { + let key = tonic::metadata::AsciiMetadataKey::from_bytes(name.as_bytes()) + .map_err(|e| Status::invalid_argument(e.to_string()))?; + let val = tonic::metadata::AsciiMetadataValue::try_from(value) + .map_err(|e| Status::invalid_argument(e.to_string()))?; + metadata.insert(key, val); + } + Ok(()) + } + fn requires_secure_transport(&self) -> bool { + self.requires_secure + } + } + + #[tokio::test] + async fn call_creds_attach_metadata() { + let addr = start_mock_server(Some("Bearer test-token")).await; + let creds = Arc::new(MockCreds { + pairs: vec![("authorization".into(), "Bearer test-token".into())], + requires_secure: false, + }); + let transport = TonicTransportBuilder::new() + .with_call_credentials(creds) + .build(&ServerConfig::new(format!("http://{addr}"))) + .await + .unwrap(); + let request = DiscoveryRequest { + type_url: "type.googleapis.com/envoy.config.listener.v3.Listener".to_string(), + ..Default::default() + }; + let request_bytes: Bytes = request.encode_to_vec().into(); + let mut stream = transport.new_stream(vec![request_bytes]).await.unwrap(); + let response = stream.recv().await.unwrap().unwrap(); + let response = DiscoveryResponse::decode(response).unwrap(); + assert_eq!(response.version_info, "1"); + } + + #[tokio::test] + async fn with_channel_secure_admits_secure_call_creds() { + let addr = start_mock_server(Some("Bearer secure-token")).await; + let channel = Endpoint::from_shared(format!("http://{addr}")) + .unwrap() + .connect_lazy(); + let creds = Arc::new(MockCreds { + pairs: vec![("authorization".into(), "Bearer secure-token".into())], + requires_secure: true, + }); + let transport = TonicTransportBuilder::new() + .with_channel(channel, true) + .with_call_credentials(creds) + .build(&ServerConfig::new(format!("http://{addr}"))) + .await + .unwrap(); + let request = DiscoveryRequest { + type_url: "type.googleapis.com/envoy.config.listener.v3.Listener".to_string(), + ..Default::default() + }; + let request_bytes: Bytes = request.encode_to_vec().into(); + let mut stream = transport.new_stream(vec![request_bytes]).await.unwrap(); + let response = stream.recv().await.unwrap().unwrap(); + let response = DiscoveryResponse::decode(response).unwrap(); + assert_eq!(response.version_info, "1"); + } + + #[tokio::test] + async fn call_creds_require_secure_transport() { + let channel = Channel::from_static("http://127.0.0.1:1").connect_lazy(); + let transport = TonicTransportBuilder::new() + .with_channel(channel, false) + .with_call_credentials(Arc::new(MockCreds { + pairs: vec![], + requires_secure: true, + })) + .build(&ServerConfig::new("http://127.0.0.1:1")) + .await + .unwrap(); + let err = transport.new_stream(vec![]).await.unwrap_err(); + assert!(matches!(err, Error::CallCredentials(_))); + } + + #[test] + fn ensure_secure_server_uri_adds_scheme_only_when_needed() { + assert_eq!( + TonicTransportBuilder::ensure_secure_server_uri( + "trafficdirector.googleapis.com:443", + true + ), + "https://trafficdirector.googleapis.com:443", + ); + assert_eq!( + TonicTransportBuilder::ensure_secure_server_uri("https://xds.example.com:443", true), + "https://xds.example.com:443" + ); + assert_eq!( + TonicTransportBuilder::ensure_secure_server_uri("unix:///etc/istio/proxy/XDS", true), + "unix:///etc/istio/proxy/XDS" + ); + assert_eq!( + TonicTransportBuilder::ensure_secure_server_uri("127.0.0.1:18000", false), + "127.0.0.1:18000" + ); + } + #[tokio::test] async fn test_tonic_transport_connect_and_stream() { - let addr = start_mock_server().await; + let addr = start_mock_server(None).await; let uri = format!("http://{addr}"); let transport = TonicTransport::connect(&uri).await.unwrap();