Add a server mode to gel-stream (#8282)

This adds a server mode to `gel-stream`, gated on a feature. We wrap up
a significant amount of network and TLS complexity behind two simple
concepts: the `Connector` and the `Acceptor`.

OpenSSL and Rustls are supported for both client and server modes. Tests
were added to ensure that they work with all combinations (which also
shook out some additional bugs). In addition, synchronous lookup of TLS
parameters by SNI is supported -- async lookup will probably be
necessary in the future but is currently not implemented yet.

In addition, `tests/certs` were regenerated as one of the certs wasn't
v3, so rustls didn't like it.

Note that client-side certificate validation is implemented, but there
is no current way to read the accepted certificate on the server side.
This will be added in a later patch.

Accept and connect have similar APIs. Both also have TLS override forms
where keys, verification modes and more may be specified.

```
    let mut acceptor = Acceptor::new_tcp(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0))
        .bind()
        .await?;
    let addr = acceptor.local_address()?;

    let accept_task = tokio::spawn(async move {
        let mut connection = acceptor.next().await.unwrap().unwrap();
        let mut buf = String::new();
        connection.read_to_string(&mut buf).await.unwrap();
        assert_eq!(buf, "Hello, world!");
    });

    let connect_task = tokio::spawn(async move {
        let target = Target::new_resolved(addr);
        let mut stm = Connector::new(target).unwrap().connect().await.unwrap();
        stm.write_all(b"Hello, world!").await?;
        Ok::<_, std::io::Error>(())
    });

```

rust/auth/src/lib.rs

@@ -34,18 +34,21 @@ pub enum AuthType {
     ScramSha256,
 }
 
-#[derive(Debug, Clone)]
+#[derive(derive_more::Debug, Clone)]
 pub enum CredentialData {
     /// A credential that always succeeds, regardless of input password. Due to
     /// the design of SCRAM-SHA-256, this cannot be used with that auth type.
     Trust,
     /// A credential that always fails, regardless of the input password.
     Deny,
     /// A plain-text password.
+    #[debug("Plain(...)")]
     Plain(String),
     /// A stored MD5 hash + salt.
+    #[debug("Md5(...)")]
     Md5(md5::StoredHash),
     /// A stored SCRAM-SHA-256 key.
+    #[debug("Scram(...)")]
     Scram(scram::StoredKey),
 }
 

rust/gel-stream/Cargo.toml

@@ -7,7 +7,9 @@ description = "A library for streaming data between clients and servers."
 [features]
 # rustls or openssl imply tokio, and tokio is the only stream we support
 # at this time.
-default = ["tokio", "rustls"]
+default = ["tokio"]
+client = []
+server = []
 tokio = ["dep:tokio"]
 rustls = ["tokio", "dep:rustls", "dep:rustls-tokio-stream", "dep:rustls-platform-verifier", "dep:webpki"]
 openssl = ["tokio", "dep:openssl", "dep:tokio-openssl", "dep:foreign-types", "dep:openssl-sys"]
@@ -17,29 +19,39 @@ __manual_tests = []
 [dependencies]
 derive_more = { version = "1", features = ["full"] }
 thiserror = "2"
-rustls-pki-types = "1"
+futures = "0.3"
 
-tokio = { version = "1", optional = true, features = ["full"] }
+# Given that this library may be used in multiple contexts, we want to limit the
+# features we enable by default.
+
+rustls-pki-types = { version = "1", default-features = false, features = ["std"] }
+
+# feature = "tokio"
+tokio = { version = "1", optional = true, default-features = false, features = ["net", "rt"] }
+hickory-resolver = { version = "0.24.2", optional = true, default-features = false, features = ["tokio-runtime", "system-config"] }
+
+# feature = "rustls"
 rustls = { version = "0.23", optional = true, default-features = false, features = ["ring", "logging", "std", "tls12"] }
-openssl = { version = "0.10.55", optional = true }
-tokio-openssl = { version = "0.6.5", optional = true }
-hickory-resolver = { version = "0.24.2", optional = true }
-rustls-tokio-stream = { version = "0.3.0", optional = true }
+rustls-tokio-stream = { version = "0.5.0", optional = true }
 rustls-platform-verifier = { version = "0.5.0", optional = true }
 webpki = { version = "0.22", optional = true }
 
+# feature = "openssl"
+openssl = { version = "0.10.55", optional = true, default-features = false }
+tokio-openssl = { version = "0.6.5", optional = true, default-features = false }
 # Get these from openssl
-foreign-types = { version = "*", optional = true }
-openssl-sys = { version = "*", optional = true }
+foreign-types = { version = "*", optional = true, default-features = false }
+openssl-sys = { version = "*", optional = true, default-features = false }
 
 [dev-dependencies]
+# Run tests with all features enabled
+gel-stream = { workspace = true, features = ["client", "server", "tokio", "rustls", "openssl"] }
+
 tokio = { version = "1", features = ["full"] }
 tempfile = "3"
 ntest = "0.9.3"
 rustls-pemfile = "2"
-
 rstest = "0.24.0"
-rustls-tokio-stream = "0.3.0"
 
 [lints]
 workspace = true

rust/gel-stream/src/client/connection.rs

@@ -1,27 +1,37 @@
+use std::marker::PhantomData;
 use std::net::SocketAddr;
 
-use super::stream::UpgradableStream;
-use super::target::{MaybeResolvedTarget, ResolvedTarget};
-use super::tokio_stream::Resolver;
-use super::{ConnectionError, Ssl, Target, TlsInit};
+use crate::common::tokio_stream::{Resolver, TokioStream};
+use crate::{ConnectionError, Ssl, StreamUpgrade, TlsDriver, UpgradableStream};
+use crate::{MaybeResolvedTarget, ResolvedTarget, Target};
 
-type Connection = UpgradableStream<super::Stream, Option<super::Ssl>>;
+type Connection<S, D> = UpgradableStream<S, D>;
 
 /// A connector can be used to connect multiple times to the same target.
-pub struct Connector {
+#[allow(private_bounds)]
+pub struct Connector<D: TlsDriver = Ssl> {
     target: Target,
     resolver: Resolver,
+    driver: PhantomData<D>,
 }
 
-impl Connector {
+impl Connector<Ssl> {
     pub fn new(target: Target) -> Result<Self, std::io::Error> {
+        Self::new_explicit(target)
+    }
+}
+
+#[allow(private_bounds)]
+impl<D: TlsDriver> Connector<D> {
+    pub fn new_explicit(target: Target) -> Result<Self, std::io::Error> {
         Ok(Self {
             target,
             resolver: Resolver::new()?,
+            driver: PhantomData,
         })
     }
 
-    pub async fn connect(&self) -> Result<Connection, ConnectionError> {
+    pub async fn connect(&self) -> Result<Connection<TokioStream, D>, ConnectionError> {
         let stream = match self.target.maybe_resolved() {
             MaybeResolvedTarget::Resolved(target) => target.connect().await?,
             MaybeResolvedTarget::Unresolved(host, port, _) => {
@@ -36,13 +46,14 @@ impl Connector {
         };
 
         if let Some(ssl) = self.target.maybe_ssl() {
-            let mut stm = UpgradableStream::new(stream, Some(Ssl::init(ssl, self.target.name())?));
+            let ssl = D::init_client(ssl, self.target.name())?;
+            let mut stm = UpgradableStream::new_client(stream, Some(ssl));
             if !self.target.is_starttls() {
                 stm.secure_upgrade().await?;
             }
             Ok(stm)
         } else {
-            Ok(UpgradableStream::new(stream, None))
+            Ok(UpgradableStream::new_client(stream, None))
         }
     }
 }