feat(grpc): Add TCP listener API in the Runtime trait + tests for server credentials (#2507)

This change adds methods to the `Runtime` trait for creating TCP
listeners and accepting remote connections. These methods are required
to test the insecure server credentials and the upcoming TLS server
credentials, which authenticate streams on the server side.

Author: arjan-bal

grpc/src/client/name_resolution/dns/test.rs

@@ -40,7 +40,7 @@ use crate::{
         },
         service_config::ServiceConfig,
     },
-    rt::{self, tokio::TokioRuntime, GrpcRuntime},
+    rt::{self, tokio::TokioRuntime, BoxFuture, GrpcRuntime, TcpOptions},
 };
 
 use super::{DnsOptions, ParseResult};
@@ -299,6 +299,14 @@ impl rt::Runtime for FakeRuntime {
     ) -> Pin<Box<dyn Future<Output = Result<Box<dyn rt::GrpcEndpoint>, String>> + Send>> {
         self.inner.tcp_stream(target, opts)
     }
+
+    fn listen_tcp(
+        &self,
+        _addr: std::net::SocketAddr,
+        _opts: TcpOptions,
+    ) -> BoxFuture<Result<Box<dyn rt::TcpListener>, String>> {
+        unimplemented!()
+    }
 }
 
 #[tokio::test]
@@ -311,7 +319,7 @@ pub(crate) async fn dns_lookup_error() {
         work_tx: work_tx.clone(),
     });
     let runtime = FakeRuntime {
-        inner: TokioRuntime {},
+        inner: TokioRuntime::default(),
         dns: FakeDns {
             latency: Duration::from_secs(0),
             lookup_result: Err("test_error".to_string()),
@@ -344,7 +352,7 @@ pub(crate) async fn dns_lookup_timeout() {
         work_tx: work_tx.clone(),
     });
     let runtime = FakeRuntime {
-        inner: TokioRuntime {},
+        inner: TokioRuntime::default(),
         dns: FakeDns {
             latency: Duration::from_secs(20),
             lookup_result: Ok(Vec::new()),

grpc/src/credentials/dyn_wrapper.rs

@@ -123,7 +123,8 @@ mod tests {
     use crate::credentials::client::ClientHandshakeInfo;
     use crate::credentials::common::{Authority, SecurityLevel};
     use crate::credentials::insecure::InsecureChannelCredentials;
-    use crate::rt::TcpOptions;
+    use crate::credentials::InsecureServerCredentials;
+    use crate::rt::{self, TcpOptions};
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio::net::TcpListener;
 
@@ -173,4 +174,49 @@ mod tests {
             .security_context()
             .validate_authority(&authority));
     }
+
+    #[tokio::test]
+    async fn test_dyn_server_credential_dispatch() {
+        let creds = InsecureServerCredentials::new();
+        let dyn_creds: Box<dyn DynServerCredentials> = Box::new(creds);
+
+        let info = dyn_creds.info();
+        assert_eq!(info.security_protocol, "insecure");
+
+        let addr = "127.0.0.1:0";
+        let runtime = rt::default_runtime();
+        let mut listener = runtime
+            .listen_tcp(addr.parse().unwrap(), TcpOptions::default())
+            .await
+            .unwrap();
+        let server_addr = *listener.local_addr();
+
+        let client_handle = tokio::spawn(async move {
+            let mut stream = tokio::net::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 (server_stream, _) = listener.accept().await.unwrap();
+
+        let result = dyn_creds.accept(server_stream, runtime).await;
+
+        assert!(result.is_ok());
+        let output = result.unwrap();
+        let mut endpoint = output.endpoint;
+        let security_info = output.security;
+
+        assert_eq!(security_info.security_protocol(), "insecure");
+        assert_eq!(security_info.security_level(), SecurityLevel::NoSecurity);
+
+        let mut buf = vec![0u8; 25];
+        endpoint.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf[..], b"hello dynamic grpc server");
+
+        client_handle.abort();
+    }
 }

grpc/src/credentials/insecure.rs

@@ -129,13 +129,17 @@ impl ServerCredentials for InsecureServerCredentials {
 #[cfg(test)]
 mod test {
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::TcpListener;
+    use tokio::net::{TcpListener, TcpStream};
 
     use crate::credentials::client::{
         ChannelCredsInternal as ClientSealed, ClientConnectionSecurityContext, ClientHandshakeInfo,
     };
     use crate::credentials::common::{Authority, SecurityLevel};
-    use crate::credentials::{ChannelCredentials, InsecureChannelCredentials};
+    use crate::credentials::server::ServerCredsInternal;
+    use crate::credentials::{
+        ChannelCredentials, InsecureChannelCredentials, InsecureServerCredentials,
+        ServerCredentials,
+    };
     use crate::rt::GrpcEndpoint;
     use crate::rt::{self, TcpOptions};
 
@@ -188,4 +192,45 @@ mod test {
             .security_context()
             .validate_authority(&authority));
     }
+
+    #[tokio::test]
+    async fn test_insecure_server_credentials() {
+        let creds = InsecureServerCredentials::new();
+
+        let info = creds.info();
+        assert_eq!(info.security_protocol, "insecure");
+
+        let addr = "127.0.0.1:0";
+        let runtime = rt::default_runtime();
+        let mut listener = runtime
+            .listen_tcp(addr.parse().unwrap(), TcpOptions::default())
+            .await
+            .unwrap();
+        let server_addr = *listener.local_addr();
+
+        let client_handle = tokio::spawn(async move {
+            let mut stream = TcpStream::connect(server_addr).await.unwrap();
+            let data = b"hello grpc";
+            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 (server_stream, _) = listener.accept().await.unwrap();
+
+        let output = creds.accept(server_stream, runtime).await.unwrap();
+        let mut endpoint = output.endpoint;
+        let security_info = output.security;
+
+        assert_eq!(security_info.security_protocol(), "insecure");
+        assert_eq!(security_info.security_level(), SecurityLevel::NoSecurity);
+
+        let mut buf = vec![0u8; 10];
+        endpoint.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf[..], b"hello grpc");
+
+        client_handle.abort();
+    }
 }

grpc/src/rt/mod.rs

@@ -31,6 +31,8 @@ pub(crate) mod tokio;
 
 pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
 pub type BoxedTaskHandle = Box<dyn TaskHandle>;
+pub type BoxEndpoint = Box<dyn GrpcEndpoint>;
+pub type ScopedBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
 
 /// An abstraction over an asynchronous runtime.
 ///
@@ -58,6 +60,13 @@ pub trait Runtime: Send + Sync + Debug {
         target: SocketAddr,
         opts: TcpOptions,
     ) -> BoxFuture<Result<Box<dyn GrpcEndpoint>, String>>;
+
+    /// Create a new listener for the given address.
+    fn listen_tcp(
+        &self,
+        addr: SocketAddr,
+        opts: TcpOptions,
+    ) -> BoxFuture<Result<Box<dyn TcpListener>, String>>;
 }
 
 /// A future that resolves after a specified duration.
@@ -122,6 +131,20 @@ impl GrpcEndpoint for Box<dyn GrpcEndpoint> {
     }
 }
 
+/// A trait representing a TCP listener capable of accepting incoming
+/// connections.
+pub trait TcpListener: Send + Sync {
+    /// Accepts a new incoming connection.
+    ///
+    /// Returns a future that resolves to a result containing the new
+    /// `GrpcEndpoint` and the remote peer's `SocketAddr`, or an error string
+    /// if acceptance fails.
+    fn accept(&mut self) -> ScopedBoxFuture<'_, Result<(BoxEndpoint, SocketAddr), String>>;
+
+    /// Returns the local socket address this listener is bound to.
+    fn local_addr(&self) -> &SocketAddr;
+}
+
 /// A fake runtime to satisfy the compiler when no runtime is enabled. This will
 ///
 /// # Panics
@@ -150,12 +173,20 @@ impl Runtime for NoOpRuntime {
     ) -> Pin<Box<dyn Future<Output = Result<Box<dyn GrpcEndpoint>, String>> + Send>> {
         unimplemented!()
     }
+
+    fn listen_tcp(
+        &self,
+        addr: SocketAddr,
+        _opts: TcpOptions,
+    ) -> BoxFuture<Result<Box<dyn TcpListener>, String>> {
+        unimplemented!()
+    }
 }
 
 pub(crate) fn default_runtime() -> GrpcRuntime {
     #[cfg(feature = "_runtime-tokio")]
     {
-        return GrpcRuntime::new(tokio::TokioRuntime {});
+        return GrpcRuntime::new(tokio::TokioRuntime::default());
     }
     #[allow(unreachable_code)]
     GrpcRuntime::new(NoOpRuntime::default())
@@ -195,4 +226,12 @@ impl GrpcRuntime {
     ) -> BoxFuture<Result<Box<dyn GrpcEndpoint>, String>> {
         self.inner.tcp_stream(target, opts)
     }
+
+    pub fn listen_tcp(
+        &self,
+        addr: SocketAddr,
+        opts: TcpOptions,
+    ) -> BoxFuture<Result<Box<dyn TcpListener>, String>> {
+        self.inner.listen_tcp(addr, opts)
+    }
 }

grpc/src/rt/tokio/mod.rs

@@ -22,29 +22,30 @@
  *
  */
 
-use std::{
-    future::Future,
-    net::{IpAddr, SocketAddr},
-    pin::Pin,
-    time::Duration,
-};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::pin::Pin;
+use std::time::Duration;
 
-use tokio::{
-    io::{AsyncRead, AsyncWrite},
-    net::TcpStream,
-    task::JoinHandle,
-};
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
 
-use crate::rt::endpoint;
+use crate::rt::{BoxEndpoint, BoxFuture, ScopedBoxFuture, TcpOptions};
 
-use super::{BoxedTaskHandle, DnsResolver, ResolverOptions, Runtime, Sleep, TaskHandle};
+use super::{
+    endpoint, BoxedTaskHandle, DnsResolver, GrpcEndpoint, ResolverOptions, Runtime, Sleep,
+    TaskHandle,
+};
 
 #[cfg(feature = "dns")]
 mod hickory_resolver;
 
 /// A DNS resolver that uses tokio::net::lookup_host for resolution. It only
 /// supports host lookups.
-struct TokioDefaultDnsResolver {}
+struct TokioDefaultDnsResolver {
+    _priv: (),
+}
 
 #[tonic::async_trait]
 impl DnsResolver for TokioDefaultDnsResolver {
@@ -66,8 +67,10 @@ impl DnsResolver for TokioDefaultDnsResolver {
     }
 }
 
-#[derive(Debug)]
-pub(crate) struct TokioRuntime {}
+#[derive(Debug, Default)]
+pub(crate) struct TokioRuntime {
+    _priv: (),
+}
 
 impl TaskHandle for JoinHandle<()> {
     fn abort(&self) {
@@ -126,14 +129,32 @@ impl Runtime for TokioRuntime {
             Ok(stream)
         })
     }
+
+    fn listen_tcp(
+        &self,
+        addr: SocketAddr,
+        _opts: TcpOptions,
+    ) -> BoxFuture<Result<Box<dyn super::TcpListener>, String>> {
+        Box::pin(async move {
+            let listener = tokio::net::TcpListener::bind(addr)
+                .await
+                .map_err(|err| err.to_string())?;
+            let local_addr = listener.local_addr().map_err(|e| e.to_string())?;
+            let listener = TokioListener {
+                inner: listener,
+                local_addr,
+            };
+            Ok(Box::new(listener) as Box<dyn super::TcpListener>)
+        })
+    }
 }
 
 impl TokioDefaultDnsResolver {
     pub fn new(opts: ResolverOptions) -> Result<Self, String> {
         if opts.server_addr.is_some() {
             return Err("Custom DNS server are not supported, enable optional feature 'dns' to enable support.".to_string());
         }
-        Ok(TokioDefaultDnsResolver {})
+        Ok(TokioDefaultDnsResolver { _priv: () })
     }
 }
 
@@ -201,13 +222,42 @@ impl super::GrpcEndpoint for TokioTcpStream {
     }
 }
 
+struct TokioListener {
+    inner: tokio::net::TcpListener,
+    local_addr: SocketAddr,
+}
+
+impl super::TcpListener for TokioListener {
+    fn accept(&mut self) -> ScopedBoxFuture<'_, Result<(BoxEndpoint, SocketAddr), String>> {
+        Box::pin(async move {
+            let (stream, addr) = self.inner.accept().await.map_err(|e| e.to_string())?;
+            Ok((
+                Box::new(TokioTcpStream {
+                    local_addr: stream
+                        .local_addr()
+                        .map_err(|err| err.to_string())?
+                        .to_string()
+                        .into_boxed_str(),
+                    peer_addr: addr.to_string().into_boxed_str(),
+                    inner: stream,
+                }) as Box<dyn GrpcEndpoint>,
+                addr,
+            ))
+        })
+    }
+
+    fn local_addr(&self) -> &SocketAddr {
+        &self.local_addr
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::{DnsResolver, ResolverOptions, Runtime, TokioDefaultDnsResolver, TokioRuntime};
 
     #[tokio::test]
     async fn lookup_hostname() {
-        let runtime = TokioRuntime {};
+        let runtime = TokioRuntime::default();
 
         let dns = runtime
             .get_dns_resolver(ResolverOptions::default())