From 1ff57ef30a5014ec8e24b5124abc59a68d20f71e Mon Sep 17 00:00:00 2001 From: Saurav Date: Wed, 1 Jul 2026 13:34:18 +0000 Subject: [PATCH] refactor(credentials): make ServerCredentials trait dyn-compatible Mirror the ChannelCredentials dyn-compat treatment from upstream PR #2703: - Remove associated type Output and generic accept - Use BoxEndpoint and #[async_trait] for object safety - Delete DynServerCredentials bridge trait (dyn_wrapper.rs) - Update LocalServerCredentials and RustlsServerCredendials impls --- grpc/src/credentials/dyn_wrapper.rs | 132 --------------------- grpc/src/credentials/local.rs | 65 ++++++++-- grpc/src/credentials/mod.rs | 15 +-- grpc/src/credentials/rustls/server/mod.rs | 13 +- grpc/src/credentials/rustls/server/test.rs | 32 +++-- grpc/src/credentials/server.rs | 5 +- 6 files changed, 93 insertions(+), 169 deletions(-) delete mode 100644 grpc/src/credentials/dyn_wrapper.rs diff --git a/grpc/src/credentials/dyn_wrapper.rs b/grpc/src/credentials/dyn_wrapper.rs deleted file mode 100644 index f6067c32e..000000000 --- a/grpc/src/credentials/dyn_wrapper.rs +++ /dev/null @@ -1,132 +0,0 @@ -/* - * - * Copyright 2026 gRPC authors. - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. - * - */ - -use tonic::async_trait; - -use crate::credentials::ProtocolInfo; -use crate::credentials::ServerCredentials; -use crate::credentials::server::HandshakeOutput as ServerHandshakeOutput; -use crate::private; -use crate::rt::BoxEndpoint; -use crate::rt::GrpcEndpoint; -use crate::rt::GrpcRuntime; -use crate::send_future::SendFuture; - -// Bridge trait for type erasure. -#[async_trait] -pub(crate) trait DynServerCredentials: Send + Sync { - async fn dyn_accept( - &self, - source: BoxEndpoint, - runtime: GrpcRuntime, - ) -> Result, String>; - - fn info(&self) -> &ProtocolInfo; -} - -#[async_trait] -impl DynServerCredentials for T -where - T: ServerCredentials, - T::Output: GrpcEndpoint, -{ - async fn dyn_accept( - &self, - source: BoxEndpoint, - runtime: GrpcRuntime, - ) -> Result, String> { - let output = SendFuture::make_send(self.accept(source, runtime, private::Internal)).await?; - Ok(ServerHandshakeOutput { - endpoint: Box::new(output.endpoint), - security: output.security, - }) - } - - fn info(&self) -> &ProtocolInfo { - self.info() - } -} - -#[cfg(test)] -mod tests { - use tokio::io::AsyncReadExt; - use tokio::io::AsyncWriteExt; - use tokio::net::TcpListener; - use tokio::net::TcpStream; - - use super::*; - use crate::credentials::LocalServerCredentials; - use crate::credentials::SecurityLevel; - use crate::rt; - use crate::rt::AsyncIoAdapter; - use crate::rt::tokio::TokioIoStream; - - #[tokio::test] - async fn test_dyn_server_credential_dispatch() { - let creds = LocalServerCredentials::new(); - let dyn_creds: Box = Box::new(creds); - - let info = dyn_creds.info(); - assert_eq!(info.security_protocol, "local"); - - let addr = "127.0.0.1:0"; - let runtime = rt::default_runtime(); - let listener = TcpListener::bind(addr).await.unwrap(); - let server_addr = listener.local_addr().unwrap(); - - let client_handle = tokio::spawn(async move { - let mut stream = TcpStream::connect(server_addr).await.unwrap(); - let data = b"hello dynamic grpc server"; - stream.write_all(data).await.unwrap(); - - // Keep the connection alive for a bit so server can read - let mut buf = vec![0u8; 1]; - let _ = stream.read(&mut buf).await; - }); - - let (stream, _) = listener.accept().await.unwrap(); - let server_stream = TokioIoStream::new_from_tcp(stream).unwrap(); - - let result = dyn_creds - .dyn_accept(Box::new(server_stream) as Box, runtime) - .await; - - assert!(result.is_ok()); - let output = result.unwrap(); - let endpoint = output.endpoint; - let security_info = output.security; - - assert_eq!(security_info.security_protocol(), "local"); - assert_eq!(security_info.security_level(), SecurityLevel::NoSecurity); - - let mut buf = vec![0u8; 25]; - AsyncIoAdapter::new(endpoint) - .read_exact(&mut buf) - .await - .unwrap(); - assert_eq!(&buf[..], b"hello dynamic grpc server"); - - client_handle.abort(); - } -} diff --git a/grpc/src/credentials/local.rs b/grpc/src/credentials/local.rs index 0a8905b5c..9f1f00328 100644 --- a/grpc/src/credentials/local.rs +++ b/grpc/src/credentials/local.rs @@ -45,7 +45,6 @@ use crate::credentials::server; use crate::credentials::server::ServerConnectionSecurityInfo; use crate::private; use crate::rt::BoxEndpoint; -use crate::rt::GrpcEndpoint; use crate::rt::GrpcRuntime; pub const PROTOCOL_NAME: &str = "local"; @@ -162,15 +161,14 @@ impl LocalServerCredentials { } } +#[async_trait] impl ServerCredentials for LocalServerCredentials { - type Output = I; - - async fn accept( + async fn accept( &self, - source: Input, + source: BoxEndpoint, _runtime: GrpcRuntime, _token: private::Internal, - ) -> Result>, String> { + ) -> Result { let security_level = security_level_for_endpoint(source.get_peer_address(), source.get_network_type())?; Ok(server::HandshakeOutput { @@ -204,7 +202,6 @@ mod test { use crate::credentials::common::Authority; use crate::rt; use crate::rt::AsyncIoAdapter; - use crate::rt::GrpcEndpoint; use crate::rt::TcpOptions; use crate::rt::tokio::TokioIoStream; @@ -321,7 +318,7 @@ mod test { let server_stream = TokioIoStream::new_from_tcp(stream).unwrap(); let output = creds - .accept(server_stream, runtime, private::Internal) + .accept(Box::new(server_stream), runtime, private::Internal) .await .unwrap(); let endpoint = output.endpoint; @@ -339,4 +336,56 @@ mod test { client_handle.abort(); } + + #[tokio::test] + async fn test_dyn_server_credential_dispatch() { + let creds = LocalServerCredentials::new(); + let dyn_creds: Arc = Arc::new(creds); + + let info = dyn_creds.info(); + assert_eq!(info.security_protocol, "local"); + + let addr = "127.0.0.1:0"; + let runtime = rt::default_runtime(); + let listener = TcpListener::bind(addr).await.unwrap(); + let server_addr = listener.local_addr().unwrap(); + + let client_handle = tokio::spawn(async move { + let mut stream = TcpStream::connect(server_addr).await.unwrap(); + let data = b"hello dynamic grpc server"; + stream.write_all(data).await.unwrap(); + + // Keep the connection alive for a bit so server can read + let mut buf = vec![0u8; 1]; + let _ = stream.read(&mut buf).await; + }); + + let (stream, _) = listener.accept().await.unwrap(); + let server_stream = TokioIoStream::new_from_tcp(stream).unwrap(); + + let result = dyn_creds + .accept( + Box::new(server_stream) as BoxEndpoint, + runtime, + private::Internal, + ) + .await; + + assert!(result.is_ok()); + let output = result.unwrap(); + let endpoint = output.endpoint; + let security_info = output.security; + + assert_eq!(security_info.security_protocol(), "local"); + assert_eq!(security_info.security_level(), SecurityLevel::NoSecurity); + + let mut buf = vec![0u8; 25]; + AsyncIoAdapter::new(endpoint) + .read_exact(&mut buf) + .await + .unwrap(); + assert_eq!(&buf[..], b"hello dynamic grpc server"); + + client_handle.abort(); + } } diff --git a/grpc/src/credentials/mod.rs b/grpc/src/credentials/mod.rs index cb5b48ee0..a01e690c6 100644 --- a/grpc/src/credentials/mod.rs +++ b/grpc/src/credentials/mod.rs @@ -37,7 +37,6 @@ pub mod call; pub(crate) mod client; -pub(crate) mod dyn_wrapper; mod local; #[cfg(feature = "tls-rustls")] pub mod rustls; @@ -56,7 +55,6 @@ use crate::credentials::client::HandshakeOutput; use crate::credentials::common::Authority; use crate::private; use crate::rt::BoxEndpoint; -use crate::rt::GrpcEndpoint; use crate::rt::GrpcRuntime; /// Client-side trait for all live gRPC wire protocols and supported transport @@ -100,11 +98,8 @@ pub trait ChannelCredentials: Send + Sync + 'static { /// Server-side trait for all live gRPC wire protocols and supported /// transport security protocols (e.g., TLS, ALTS). -#[trait_variant::make(Send)] -pub trait ServerCredentials: Sync + 'static { - #[doc(hidden)] - type Output; - +#[async_trait] +pub trait ServerCredentials: Send + Sync + 'static { /// Provides the ProtocolInfo of these credentials. fn info(&self) -> &ProtocolInfo; @@ -113,12 +108,12 @@ pub trait ServerCredentials: Sync + 'static { /// This method wraps the incoming raw `source` connection with the configured /// security protocol (e.g., TLS). #[doc(hidden)] - async fn accept( + async fn accept( &self, - source: Input, + source: BoxEndpoint, runtime: GrpcRuntime, token: private::Internal, - ) -> Result>, String>; + ) -> Result; } /// Defines the level of protection provided by an established connection. diff --git a/grpc/src/credentials/rustls/server/mod.rs b/grpc/src/credentials/rustls/server/mod.rs index d571545df..45fefa8f7 100644 --- a/grpc/src/credentials/rustls/server/mod.rs +++ b/grpc/src/credentials/rustls/server/mod.rs @@ -56,7 +56,7 @@ use crate::credentials::server::HandshakeOutput; use crate::credentials::server::ServerConnectionSecurityInfo; use crate::private; use crate::rt::AsyncIoAdapter; -use crate::rt::GrpcEndpoint; +use crate::rt::BoxEndpoint; use crate::rt::GrpcRuntime; #[cfg(test)] @@ -338,15 +338,14 @@ impl ProducesTickets for NoTicketer { } } +#[tonic::async_trait] impl ServerCredentials for RustlsServerCredendials { - type Output = TlsStream; - - async fn accept( + async fn accept( &self, - source: Input, + source: BoxEndpoint, _runtime: GrpcRuntime, _token: private::Internal, - ) -> Result>, String> { + ) -> Result { let input_io = AsyncIoAdapter::new(source); let tls_stream = self .acceptor @@ -366,7 +365,7 @@ impl ServerCredentials for RustlsServerCredendials { ); let endpoint = TlsStream::new(RustlsStream::Server(tls_stream)); Ok(HandshakeOutput { - endpoint, + endpoint: Box::new(endpoint), security: auth_info, }) } diff --git a/grpc/src/credentials/rustls/server/test.rs b/grpc/src/credentials/rustls/server/test.rs index 988845565..9c6dc2af9 100644 --- a/grpc/src/credentials/rustls/server/test.rs +++ b/grpc/src/credentials/rustls/server/test.rs @@ -74,7 +74,9 @@ async fn test_tls_server_handshake() { let server_task = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); - let result = creds.accept(stream, runtime, private::Internal).await; + let result = creds + .accept(Box::new(stream), runtime, private::Internal) + .await; assert!( result.is_ok(), "Server handshake failed: {:?}", @@ -130,7 +132,9 @@ async fn test_tls_server_handshake_no_alpn() { let server_task = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); - let result = creds.accept(stream, runtime, private::Internal).await; + let result = creds + .accept(Box::new(stream), runtime, private::Internal) + .await; assert!(result.is_err(), "Server handshake should have failed"); }); @@ -174,7 +178,9 @@ async fn test_tls_server_handshake_bad_alpn() { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let runtime = rt::default_runtime(); - let result = creds.accept(stream, runtime, private::Internal).await; + let result = creds + .accept(Box::new(stream), runtime, private::Internal) + .await; assert!(result.is_err(), "Server handshake should have failed"); }); @@ -212,7 +218,7 @@ async fn test_tls_handshake_alpn_h1_and_h2() { let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let runtime = rt::default_runtime(); creds - .accept(stream, runtime, private::Internal) + .accept(Box::new(stream), runtime, private::Internal) .await .unwrap(); }); @@ -257,7 +263,9 @@ async fn test_tls_server_mtls_require_fail() { let server_task = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); - let result = creds.accept(stream, runtime, private::Internal).await; + let result = creds + .accept(Box::new(stream), runtime, private::Internal) + .await; assert!(result.is_err(), "Handshake should fail without client cert"); }); @@ -309,7 +317,7 @@ async fn test_tls_server_mtls_success() { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let result = creds - .accept(stream, runtime, private::Internal) + .accept(Box::new(stream), runtime, private::Internal) .await .expect("Server handshake failed"); let mut stream = AsyncIoAdapter::new(result.endpoint); @@ -369,7 +377,7 @@ async fn test_tls_server_mtls_optional() { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let result = creds - .accept(stream, runtime, private::Internal) + .accept(Box::new(stream), runtime, private::Internal) .await .expect("Server handshake failed"); let mut stream = AsyncIoAdapter::new(result.endpoint); @@ -419,7 +427,7 @@ async fn test_tls_server_key_log() { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let result = creds - .accept(stream, runtime, private::Internal) + .accept(Box::new(stream), runtime, private::Internal) .await .expect("Server handshake failed"); let mut stream = AsyncIoAdapter::new(result.endpoint); @@ -473,7 +481,9 @@ async fn check_resumption_disabled(versions: Vec<&'static rustls::SupportedProto let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let runtime = rt::default_runtime(); - let result = creds.accept(stream, runtime, private::Internal).await; + let result = creds + .accept(Box::new(stream), runtime, private::Internal) + .await; assert!(result.is_ok()); let stream = result.unwrap().endpoint; AsyncIoAdapter::new(stream) @@ -549,7 +559,9 @@ async fn test_tls_server_sni() { let (stream, _) = listener.accept().await.unwrap(); let stream = TokioIoStream::new_from_tcp(stream).unwrap(); let runtime = rt::default_runtime(); - let result = creds.accept(stream, runtime, private::Internal).await; + let result = creds + .accept(Box::new(stream), runtime, private::Internal) + .await; assert!( result.is_ok(), "Server handshake failed: {:?}", diff --git a/grpc/src/credentials/server.rs b/grpc/src/credentials/server.rs index 7eeb7858a..9c9591616 100644 --- a/grpc/src/credentials/server.rs +++ b/grpc/src/credentials/server.rs @@ -24,9 +24,10 @@ use crate::attributes::Attributes; use crate::credentials::SecurityLevel; +use crate::rt::BoxEndpoint; -pub struct HandshakeOutput { - pub endpoint: T, +pub struct HandshakeOutput { + pub endpoint: BoxEndpoint, pub security: ServerConnectionSecurityInfo, }