Add TLS support

[adwk67]

tokio-zookeeper does not currently support TLS: this needs to be added before we can support Zookeeper 3.10 (or patch ZOOKEEPER-4276). Some of the unit tests fail and the code compiles with a warning. Recent additions (for release 25.11) are in main and need to part of a new release 0.5.0.

[lfrancke]

I understood that this is one option. Another option would be to switch to a different library, correct?
I'd like to have that clarified during a refinement.

[maltesander]

Research

I had a look at at "our" tokio-zookeeper and 2 alternatives. Starting with the alternatives:

rust-zookeeper

Repo: https://github.com/bonifaido/rust-zookeeper
Last updated: 3 Years ago (code, dependencies)
License: MIT

No TLS support. No real maintenance. Does not make sense.

zookeeper-client-rust

Repo: https://github.com/kezhuw/zookeeper-client-rust/
Last updated: 6 Months ago
License: MIT

Features:

• TLS support.
• FIPS mode support with compile-time and runtime toggle.
• SASL support.
• Async runtime agnostic.
• Curator compatible lock recipes.
• Versatile transactional like APIs LockClient, MultiWriter, MultiReader and CheckWriter.
• Up to date API set comparing to official Java client (-> this means it is way different than the java client API)

While it has a couple of nice features, the different API approach will result in more changes regarding the znode_controller in the operator.

tokio-zookeeper

Repo: https://github.com/stackabletech/tokio-zookeeper/

No TLS support. BUT: We have control over it.

Features:

• Close to the Java client API
• Implemented and running with zk operator (least possible breakage)

Outcome

In my opinion, we only have one alternative (zookeeper-client-rust), it already supports TLS but we lose control and it requires changing the znode_controller.
Using "our" tokio-zookeeper, we should be able to TLSify it with similar effort.

This is a quick and dirty spike (there are a couple of more changes, just to show the trait)

pub trait ZooKeeperTransport: AsyncRead + AsyncWrite + Sized + Send + 'static {
    type Addr: Send + Clone;
    type Hostname: Send + Clone; // added for TLS -> should be ClientConfig or custom TlsConfig
    type ConnectError: Error + Send + Sync + 'static;
    fn connect(
        addr: Self::Addr,
        host: Self::Hostname,
    ) -> impl std::future::Future<Output = Result<Self, Self::ConnectError>> + std::marker::Send;
}

// EXISTING IMPL
impl ZooKeeperTransport for tokio::net::TcpStream {
    type Addr = SocketAddr;
    type Hostname = (); // not required for plain
    type ConnectError = tokio::io::Error;
    async fn connect(addr: Self::Addr, _host: Self::Hostname) -> Result<Self, Self::ConnectError> {
        tokio::net::TcpStream::connect(addr).await
    }
}

// NEW TLS IMPL
impl ZooKeeperTransport for tokio_rustls::client::TlsStream<tokio::net::TcpStream> {
    type Addr = SocketAddr;
    type Hostname = String;
    type ConnectError = tokio::io::Error;

    async fn connect(addr: Self::Addr, host: Self::Hostname) -> Result<Self, Self::ConnectError> {
        let root_store = rustls::RootCertStore::empty();
        // Should be passed through a tls config
        let config = rustls::ClientConfig::builder()
            .with_root_certificates(root_store)
            .with_no_client_auth();

        let connector = TlsConnector::from(Arc::new(config));

        let domain = rustls_pki_types::ServerName::try_from(host)
            .map_err(|e| tokio::io::Error::new(ErrorKind::InvalidInput, e))?;

        let stream = TcpStream::connect(addr).await?;

        connector.connect(domain, stream).await
    }
}

This could be implemented much nicer, e.g. enum for plain or tls stream.

Problem

Independent of the choice of client, we have a problem getting the certs for TLS into the operator.

One operator manages multiple zookeeper clusters, of which different clusters may use different SecretClasses.
How can we make sure the operator can communicate with each cluster using its SecretClass? We also support mTLS (client auth) for zookeeper authentication.

Mounting anything is not an option since the secretclasses are only known at runtime. Reading the secret content sounds wrong.

This requires refinement.