Provide an example on how to use rustls::server::Acceptor

[markandpathak]

Hello,

I am new to both Rust and async in Rust with tokio so there might be something I might have missed or don't know. Rustls recommends here to use rustls::server::Acceptor for retrieving server certificates dynamically based on ClientHello when working in async manner. But I could not find any useful information on how to use that with Tokio-rustls.

What I have observed is Acceptor implements read_tls method which accepts a parameter which implements Read trait. But tokio::net::TcpStream implements AsyncRead which is not compatible to be passed in Acceptor. Or may be I don't know how to use that.

Any guidance or minimal example is appreciated on how to use Acceptor.

If there are any better ways to implement dynamic server certificates fetched from a key store, I would like to know that too.

PS: I need TCP socket stream directly as I am implementing a Pub/Sub Broker both for experimentation and learning Rust for real use case.

[ctz]

You will want to start at https://docs.rs/tokio-rustls/latest/tokio_rustls/struct.LazyConfigAcceptor.html rather than rustls::server::Acceptor.

[markandpathak]

Thank you very much! I have a lot more clarity now. I guess these basic examples should be included in the examples folder. It can save a lot of time for other developers figuring out these fairly common functionalities. I guess following examples needs to be added. I'll try to create some examples and submit through PR. I'll need some guidance though:

Client Side:

• Client Cert and Cipher suite selection
• Server cert verification though inbuilt verifier and external verification method

Server Side:

• SNI based server cert selection
• Server cert selection based on cipher suits provided by client (ECC/RSA etc)

[djc]

I'm generally wary of adding a bunch of examples that are all slightly different (but will usually also share a lot of code), but I agree that it makes sense to add an example for the LazyConfigAcceptor. Since most of these other things are not specific to tokio-rustls, it might make sense to defer to examples/documentation in the rustls repo instead.

[markandpathak]

I just had one follow up question: How do you define a timeout on client-hello and overall TLS handshake? How to prevent attacks like Slowloris? Where to configure it? I don't see any configuration in server.rs example as well.

[djc]

We have #4 for the general Acceptor case. For LazyConfigAcceptor you'll want to wrap the acceptor in something like tokio::time::timeout() for now.

[aidant]

I've been trying to achieve the same and was quite pleased when I discovered this issue calling out the LazyConfigAcceptor. Since initially I was thinking of using axum_server but was dissapointed when I saw it doesn't currently support an async SNI callback. If its helpful to anyone else, I've got an initial prototype working based on the example for the LazyConfigAcceptor and the axum::serve function. I've not implemented the timeout, should it be for acceptor.as_mut().await or start.into_stream(config).await or both?

Note I've generated the certs with mkcert:

mkcert -cert-file localhost-cert.pem -key-file localhost-key.pem "localhost" "127.0.0.1" "::1"

use axum::{extract::connect_info::IntoMakeServiceWithConnectInfo, Router};
use hyper_util::{
    rt::{TokioExecutor, TokioIo},
    server::conn::auto::Builder,
    service::TowerToHyperService,
};
use std::{io, net::SocketAddr, time::Duration};
use tokio::{
    io::AsyncWriteExt,
    net::{TcpListener, TcpStream},
};
use tokio_rustls::{rustls::server::Acceptor, LazyConfigAcceptor};
use tower_service::Service;
use tracing::{error, trace};

pub async fn serve_https(
    tcp_listener: TcpListener,
    mut make_service: IntoMakeServiceWithConnectInfo<Router, SocketAddr>,
) -> Result<(), anyhow::Error> {
    loop {
        let (tcp_stream, remote_addr) = match tcp_accept(&tcp_listener).await {
            Some(conn) => conn,
            None => continue,
        };

        let tower_service = make_service
            .call(remote_addr)
            .await
            .unwrap_or_else(|err| match err {});

        let hyper_service = TowerToHyperService::new(tower_service);
        let acceptor = LazyConfigAcceptor::new(Acceptor::default(), tcp_stream);

        tokio::spawn(async move {
            tokio::pin!(acceptor);

            let result: Result<(), anyhow::Error> = async {
                let start = acceptor.as_mut().await?;

                let client_hello = start.client_hello();
                let config = get_server_config(client_hello).await;
                let tcp_stream = start.into_stream(config).await?;

                trace!("connection {remote_addr} accepted");

                let tcp_stream = TokioIo::new(tcp_stream);

                Builder::new(TokioExecutor::new())
                    .serve_connection_with_upgrades(tcp_stream, hyper_service)
                    .await
                    .or_else(|_err| Ok(()))
            }
            .await;

            if let Err(err) = result {
                error!("unable to process request: {err}");

                if let Some(mut stream) = acceptor.take_io() {
                    stream
                        .write_all("HTTP/1.1 400 Bad Request\r\n\r\n\r\n".as_bytes())
                        .await
                        .unwrap_or_else(|err| {
                            error!("unable to send error response: {err}");
                        });
                }
            }
        });
    }
}

fn is_connection_error(e: &io::Error) -> bool {
    matches!(
        e.kind(),
        io::ErrorKind::ConnectionRefused
            | io::ErrorKind::ConnectionAborted
            | io::ErrorKind::ConnectionReset
    )
}

async fn tcp_accept(listener: &TcpListener) -> Option<(TcpStream, SocketAddr)> {
    match listener.accept().await {
        Ok(conn) => Some(conn),
        Err(e) => {
            if is_connection_error(&e) {
                return None;
            }

            error!("accept error: {e}");
            tokio::time::sleep(Duration::from_secs(1)).await;
            None
        }
    }
}

pub async fn get_server_config(client_hello: ClientHello<'_>) -> Arc<ServerConfig> {
    let mut config = ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(
            CertificateDer::pem_file_iter(
                PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("localhost-cert.pem"),
            )
            .unwrap()
            .map(|cert| cert.unwrap())
            .collect(),
            PrivateKeyDer::from_pem_file(
                PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("localhost-key.pem"),
            )
            .unwrap(),
        )
        .unwrap();

    config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];

    Arc::new(config)
}

[djc]

I've not implemented the timeout, should it be for acceptor.as_mut().await or start.into_stream(config).await or both?

You'll want both: the former covers receiving the ClientHello, and the latter covers the rest of the TLS handshake.

I started on a fairly unopinionated server implementation that could maybe live in hyper-rustls, I should finish it up some time.

[markandpathak]

I think it might go little off topic, but I want to implement this case: In a pub/sub broker it is inefficient to allocate a buffer for each connection as there can be potentially thousands of connections which remain idle most of the time. This basically consumes too much memory. So there can be thousands of clients, but at any instance only hundreds are sending data.

I see that underlying tcpstream has readable method which basically wait until socket becomes readable. Once readable, I am thinking to pass on message through a channel to some worker task which will read the socket by fetching stream from hashmap and calling poll_read into a buffer and process the incoming data with it's own pre-allocated buffer. I am trying like this:

match acceptor.as_mut().await {
            Ok(start) => {
                let client_hello = start.client_hello();
                // Get the dynamic config here
                let stream = start.into_stream(final_config).await.unwrap();
                // Store the stream into a HashMap of struct along with other client related info
                tokio::spawn(async move {
                    let client_ctx_data = client_ctx;
                    let (tcp_stream, conn_data) = stream.into_inner();
                    tcp_stream.readable().await?;
                    // Send message over a channel to worker task that this connection is trying to send something
                })

My question is: Is this right way to do this? or tokio-rustls has some method to wait until stream is readable without copying anything in a buffer?

[djc]

In async Rust, tasks are generally only woken up by the runtime when there is something for them to do, so that doesn't seem like a process you should try to duplicate at your level?

[markandpathak]

I agree but here goal is to delay the read buffer allocation until actually needed. Even though execution wise they are same, they wake up when there's data, there's a significant difference between implementing with readable and read method.

When using read I need to allocate buffer beforehand. Imagine 100k connections and I am allocating 4kb for each connection, with read I need to allocate ~400Mb just as buffer which will remain un-utilized for most of the connections for majority of time.

In readable I wait until client actually sends data after which I allocate the buffer. Furthermore from 100k connections only few thousand will be sending data at the same time, i'll be allocating buffer for those connections only which are readable. You can imagine the memory optimization that can be achieved.

[djc]

I think you should be able to use something like get_ref().readable().await for most tokio-rustls types.