Add preliminary rustls support

Co-authored-by: Harald Gutmann <[email protected]>

17 files changed, 1299 insertions, 17 deletions

diff --git a/.bleep b/.bleep
index 29706e1..44a497b 100644
--- a/.bleep
+++ b/.bleep
@@ -1 +1 @@
-46b4788e5c9c00dfaa327800b22cb84c8f695b5c
\ No newline at end of file
+3919de32fe7a184847dd6ce7da98247ec9e1eb86
\ No newline at end of file
diff --git a/pingora-core/src/connectors/tls/mod.rs b/pingora-core/src/connectors/tls/mod.rs
index e250418..25a7b48 100644
--- a/pingora-core/src/connectors/tls/mod.rs
+++ b/pingora-core/src/connectors/tls/mod.rs
@@ -18,6 +18,12 @@ mod boringssl_openssl;
 #[cfg(feature = "openssl_derived")]
 pub use boringssl_openssl::*;
 
+#[cfg(feature = "rustls")]
+mod rustls;
+
+#[cfg(feature = "rustls")]
+pub use rustls::*;
+
 ///    OpenSSL considers underscores in hostnames non-compliant.
 ///    We replace the underscore in the leftmost label as we must support these
 ///    hostnames for wildcard matches and we have not patched OpenSSL.
diff --git a/pingora-core/src/connectors/tls/rustls/mod.rs b/pingora-core/src/connectors/tls/rustls/mod.rs
new file mode 100644
index 0000000..7d10c0e
--- /dev/null
+++ b/pingora-core/src/connectors/tls/rustls/mod.rs
@@ -0,0 +1,225 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use log::debug;
+use pingora_error::{
+    Error,
+    ErrorType::{ConnectTimedout, InvalidCert},
+    OrErr, Result,
+};
+use pingora_rustls::{
+    load_ca_file_into_store, load_certs_key_file, load_platform_certs_incl_env_into_store, version,
+    CertificateDer, ClientConfig as RusTlsClientConfig, PrivateKeyDer, RootCertStore,
+    TlsConnector as RusTlsConnector,
+};
+
+use crate::protocols::tls::{client::handshake, TlsStream};
+use crate::{connectors::ConnectorOptions, listeners::ALPN, protocols::IO, upstreams::peer::Peer};
+
+use super::replace_leftmost_underscore;
+
+#[derive(Clone)]
+pub struct Connector {
+    pub ctx: Arc<TlsConnector>,
+}
+
+impl Connector {
+    pub fn new(config_opt: Option<ConnectorOptions>) -> Self {
+        TlsConnector::build_connector(config_opt)
+    }
+}
+
+pub(crate) struct TlsConnector {
+    config: RusTlsClientConfig,
+    ca_certs: RootCertStore,
+}
+
+impl TlsConnector {
+    pub(crate) fn build_connector(options: Option<ConnectorOptions>) -> Connector
+    where
+        Self: Sized,
+    {
+        // NOTE: Rustls only supports TLS 1.2 & 1.3
+
+        // TODO: currently using Rustls defaults
+        // - support SSLKEYLOGFILE
+        // - set supported ciphers/algorithms/curves
+        // - add options for CRL/OCSP validation
+
+        let (ca_certs, certs_key) = {
+            let mut ca_certs = RootCertStore::empty();
+            let mut certs_key = None;
+
+            if let Some(conf) = options.as_ref() {
+                if let Some(ca_file_path) = conf.ca_file.as_ref() {
+                    load_ca_file_into_store(ca_file_path, &mut ca_certs);
+                } else {
+                    load_platform_certs_incl_env_into_store(&mut ca_certs);
+                }
+                if let Some((cert, key)) = conf.cert_key_file.as_ref() {
+                    certs_key = load_certs_key_file(cert, key);
+                }
+                // TODO: support SSLKEYLOGFILE
+            } else {
+                load_platform_certs_incl_env_into_store(&mut ca_certs);
+            }
+
+            (ca_certs, certs_key)
+        };
+
+        // TODO: WebPkiServerVerifier for CRL/OCSP validation
+        let builder =
+            RusTlsClientConfig::builder_with_protocol_versions(&[&version::TLS12, &version::TLS13])
+                .with_root_certificates(ca_certs.clone());
+
+        let config = match certs_key {
+            Some((certs, key)) => {
+                match builder.with_client_auth_cert(certs.clone(), key.clone_key()) {
+                    Ok(config) => config,
+                    Err(err) => {
+                        // TODO: is there a viable alternative to the panic?
+                        // falling back to no client auth... does not seem to be reasonable.
+                        panic!(
+                            "{}",
+                            format!("Failed to configure client auth cert/key. Error: {}", err)
+                        );
+                    }
+                }
+            }
+            None => builder.with_no_client_auth(),
+        };
+
+        Connector {
+            ctx: Arc::new(TlsConnector { config, ca_certs }),
+        }
+    }
+}
+
+pub async fn connect<T, P>(
+    stream: T,
+    peer: &P,
+    alpn_override: Option<ALPN>,
+    tls_ctx: &TlsConnector,
+) -> Result<TlsStream<T>>
+where
+    T: IO,
+    P: Peer + Send + Sync,
+{
+    let mut config = tls_ctx.config.clone();
+
+    // TODO: setup CA/verify cert store from peer
+    // looks like the fields are always None
+    // peer.get_ca()
+
+    let key_pair = peer.get_client_cert_key();
+    let updated_config: Option<RusTlsClientConfig> = match key_pair {
+        None => None,
+        Some(key_arc) => {
+            debug!("setting client cert and key");
+
+            let mut cert_chain = vec![];
+            debug!("adding leaf certificate to mTLS cert chain");
+            cert_chain.push(key_arc.leaf().to_owned());
+
+            debug!("adding intermediate certificates to mTLS cert chain");
+            key_arc
+                .intermediates()
+                .to_owned()
+                .iter()
+                .map(|i| i.to_vec())
+                .for_each(|i| cert_chain.push(i));
+
+            let certs: Vec<CertificateDer> = cert_chain
+                .into_iter()
+                .map(|c| c.as_slice().to_owned().into())
+                .collect();
+            let private_key: PrivateKeyDer =
+                key_arc.key().as_slice().to_owned().try_into().unwrap();
+
+            let builder = RusTlsClientConfig::builder_with_protocol_versions(&[
+                &version::TLS12,
+                &version::TLS13,
+            ])
+            .with_root_certificates(tls_ctx.ca_certs.clone());
+
+            let updated_config = builder
+                .with_client_auth_cert(certs, private_key)
+                .explain_err(InvalidCert, |e| {
+                    format!(
+                        "Failed to use peer cert/key to update Rustls config: {:?}",
+                        e
+                    )
+                })?;
+            Some(updated_config)
+        }
+    };
+
+    if let Some(alpn) = alpn_override.as_ref().or(peer.get_alpn()) {
+        config.alpn_protocols = alpn.to_wire_protocols();
+    }
+
+    // TODO: curve setup from peer
+    // - second key share from peer, currently only used in boringssl with PQ features
+
+    let tls_conn = if let Some(cfg) = updated_config {
+        RusTlsConnector::from(Arc::new(cfg))
+    } else {
+        RusTlsConnector::from(Arc::new(config))
+    };
+
+    // TODO: for consistent behaviour between TLS providers some additions are required
+    // - allowing to disable verification
+    // - the validation/replace logic would need adjustments to match the boringssl/openssl behaviour
+    //   implementing a custom certificate_verifier could be used to achieve matching behaviour
+    //let d_conf = config.dangerous();
+    //d_conf.set_certificate_verifier(...);
+
+    let mut domain = peer.sni().to_string();
+    if peer.sni().is_empty() {
+        // use ip in case SNI is not present
+        // TODO: disable validation
+        domain = peer.address().as_inet().unwrap().ip().to_string()
+    }
+
+    if peer.verify_cert() && peer.verify_hostname() {
+        // TODO: streamline logic with replacing first underscore within TLS implementations
+        if let Some(sni_s) = replace_leftmost_underscore(peer.sni()) {
+            domain = sni_s;
+        }
+        if let Some(alt_cn) = peer.alternative_cn() {
+            if !alt_cn.is_empty() {
+                domain = alt_cn.to_string();
+                // TODO: streamline logic with replacing first underscore within TLS implementations
+                if let Some(alt_cn_s) = replace_leftmost_underscore(alt_cn) {
+                    domain = alt_cn_s;
+                }
+            }
+        }
+    }
+
+    let connect_future = handshake(&tls_conn, &domain, stream);
+
+    match peer.connection_timeout() {
+        Some(t) => match pingora_timeout::timeout(t, connect_future).await {
+            Ok(res) => res,
+            Err(_) => Error::e_explain(
+                ConnectTimedout,
+                format!("connecting to server {}, timeout {:?}", peer, t),
+            ),
+        },
+        None => connect_future.await,
+    }
+}
diff --git a/pingora-core/src/listeners/mod.rs b/pingora-core/src/listeners/mod.rs
index 0c5d78e..09dc606 100644
--- a/pingora-core/src/listeners/mod.rs
+++ b/pingora-core/src/listeners/mod.rs
@@ -44,6 +44,7 @@ pub trait TlsAccept {
     /// This function is called in the middle of a TLS handshake. Structs who
     /// implement this function should provide tls certificate and key to the
     /// [TlsRef] via `ssl_use_certificate` and `ssl_use_private_key`.
+    /// Note. This is only supported for openssl and boringssl
     async fn certificate_callback(&self, _ssl: &mut TlsRef) -> () {
         // does nothing by default
     }
diff --git a/pingora-core/src/listeners/tls/mod.rs b/pingora-core/src/listeners/tls/mod.rs
index 0bcaeac..4f59fcb 100644
--- a/pingora-core/src/listeners/tls/mod.rs
+++ b/pingora-core/src/listeners/tls/mod.rs
@@ -17,3 +17,9 @@ mod boringssl_openssl;
 
 #[cfg(feature = "openssl_derived")]
 pub use boringssl_openssl::*;
+
+#[cfg(feature = "rustls")]
+mod rustls;
+
+#[cfg(feature = "rustls")]
+pub use rustls::*;
diff --git a/pingora-core/src/listeners/tls/rustls/mod.rs b/pingora-core/src/listeners/tls/rustls/mod.rs
new file mode 100644
index 0000000..cfb2811
--- /dev/null
+++ b/pingora-core/src/listeners/tls/rustls/mod.rs
@@ -0,0 +1,116 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use crate::listeners::TlsAcceptCallbacks;
+use crate::protocols::tls::{server::handshake, server::handshake_with_callback, TlsStream};
+use log::debug;
+use pingora_error::ErrorType::InternalError;
+use pingora_error::{Error, ErrorSource, ImmutStr, OrErr, Result};
+use pingora_rustls::load_certs_key_file;
+use pingora_rustls::ServerConfig;
+use pingora_rustls::{version, TlsAcceptor as RusTlsAcceptor};
+
+use crate::protocols::{ALPN, IO};
+
+/// The TLS settings of a listening endpoint
+pub struct TlsSettings {
+    alpn_protocols: Option<Vec<Vec<u8>>>,
+    cert_path: String,
+    key_path: String,
+}
+
+pub struct Acceptor {
+    pub acceptor: RusTlsAcceptor,
+    callbacks: Option<TlsAcceptCallbacks>,
+}
+
+impl TlsSettings {
+    pub fn build(self) -> Acceptor {
+        let (certs, key) =
+            load_certs_key_file(&self.cert_path, &self.key_path).unwrap_or_else(|| {
+                panic!(
+                    "Failed to load provided certificates \"{}\" or key \"{}\".",
+                    self.cert_path, self.key_path
+                )
+            });
+
+        let mut config =
+            ServerConfig::builder_with_protocol_versions(&[&version::TLS12, &version::TLS13])
+                .with_no_client_auth()
+                .with_single_cert(certs, key)
+                .explain_err(InternalError, |e| {
+                    format!("Failed to create server listener config: {}", e)
+                })
+                .unwrap();
+
+        if let Some(alpn_protocols) = self.alpn_protocols {
+            config.alpn_protocols = alpn_protocols;
+        }
+
+        Acceptor {
+            acceptor: RusTlsAcceptor::from(Arc::new(config)),
+            callbacks: None,
+        }
+    }
+
+    /// Enable HTTP/2 support for this endpoint, which is default off.
+    /// This effectively sets the ALPN to prefer HTTP/2 with HTTP/1.1 allowed
+    pub fn enable_h2(&mut self) {
+        self.set_alpn(ALPN::H2H1);
+    }
+
+    fn set_alpn(&mut self, alpn: ALPN) {
+        self.alpn_protocols = Some(alpn.to_wire_protocols());
+    }
+
+    pub fn intermediate(cert_path: &str, key_path: &str) -> Result<Self>
+    where
+        Self: Sized,
+    {
+        Ok(TlsSettings {
+            alpn_protocols: None,
+            cert_path: cert_path.to_string(),
+            key_path: key_path.to_string(),
+        })
+    }
+
+    pub fn with_callbacks() -> Result<Self>
+    where
+        Self: Sized,
+    {
+        // TODO: verify if/how callback in handshake can be done using Rustls
+        Err(Error::create(
+            InternalError,
+            ErrorSource::Internal,
+            Some(ImmutStr::from(
+                "Certificate callbacks are not supported with feature \"rustls\".",
+            )),
+            None,
+        ))
+    }
+}
+
+impl Acceptor {
+    pub async fn tls_handshake<S: IO>(&self, stream: S) -> Result<TlsStream<S>> {
+        debug!("new tls session");
+        // TODO: be able to offload this handshake in a thread pool
+        if let Some(cb) = self.callbacks.as_ref() {
+            handshake_with_callback(self, stream, cb).await
+        } else {
+            handshake(self, stream).await
+        }
+    }
+}
diff --git a/pingora-core/src/protocols/tls/mod.rs b/pingora-core/src/protocols/tls/mod.rs
index c479a2d..04e8fd1 100644
--- a/pingora-core/src/protocols/tls/mod.rs
+++ b/pingora-core/src/protocols/tls/mod.rs
@@ -23,6 +23,12 @@ mod boringssl_openssl;
 #[cfg(feature = "openssl_derived")]
 pub use boringssl_openssl::*;
 
+#[cfg(feature = "rustls")]
+mod rustls;
+
+#[cfg(feature = "rustls")]
+pub use rustls::*;
+
 #[cfg(not(feature = "any_tls"))]
 pub mod noop_tls;
 
diff --git a/pingora-core/src/protocols/tls/rustls/client.rs b/pingora-core/src/protocols/tls/rustls/client.rs
new file mode 100644
index 0000000..bb9edc2
--- /dev/null
+++ b/pingora-core/src/protocols/tls/rustls/client.rs
@@ -0,0 +1,43 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Rustls TLS client specific implementation
+
+use crate::protocols::tls::rustls::TlsStream;
+use crate::protocols::IO;
+use pingora_error::ErrorType::TLSHandshakeFailure;
+use pingora_error::{Error, OrErr, Result};
+use pingora_rustls::TlsConnector;
+
+// Perform the TLS handshake for the given connection with the given configuration
+pub async fn handshake<S: IO>(
+    connector: &TlsConnector,
+    domain: &str,
+    io: S,
+) -> Result<TlsStream<S>> {
+    let mut stream = TlsStream::from_connector(connector, domain, io)
+        .await
+        .explain_err(TLSHandshakeFailure, |e| {
+            format!("tip: tls stream error: {e}")
+        })?;
+
+    let handshake_result = stream.connect().await;
+    match handshake_result {
+        Ok(()) => Ok(stream),
+        Err(e) => {
+            let context = format!("TLS connect() failed: {e}, SNI: {domain}");
+            Error::e_explain(TLSHandshakeFailure, context)
+        }
+    }
+}
diff --git a/pingora-core/src/protocols/tls/rustls/mod.rs b/pingora-core/src/protocols/tls/rustls/mod.rs
new file mode 100644
index 0000000..9622fac
--- /dev/null
+++ b/pingora-core/src/protocols/tls/rustls/mod.rs
@@ -0,0 +1,36 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod client;
+pub mod server;
+mod stream;
+
+pub use stream::*;
+use x509_parser::prelude::FromDer;
+
+pub type CaType = [Box<CertWrapper>];
+
+#[derive(Debug)]
+#[repr(transparent)]
+pub struct CertWrapper(pub [u8]);
+
+impl CertWrapper {
+    pub fn not_after(&self) -> String {
+        let (_, x509cert) = x509_parser::certificate::X509Certificate::from_der(&self.0)
+            .expect("Failed to parse certificate from DER format.");
+        x509cert.validity.not_after.to_string()
+    }
+}
+
+pub struct TlsRef;
diff --git a/pingora-core/src/protocols/tls/rustls/server.rs b/pingora-core/src/protocols/tls/rustls/server.rs
new file mode 100644
index 0000000..49f303f
--- /dev/null
+++ b/pingora-core/src/protocols/tls/rustls/server.rs
@@ -0,0 +1,108 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Rustls TLS server specific implementation
+
+use crate::listeners::TlsAcceptCallbacks;
+use crate::protocols::tls::rustls::TlsStream;
+use crate::protocols::IO;
+use crate::{listeners::tls::Acceptor, protocols::Shutdown};
+use async_trait::async_trait;
+use log::warn;
+use pingora_error::{ErrorType::*, OrErr, Result};
+use std::pin::Pin;
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
+
+impl<S: AsyncRead + AsyncWrite + Send + Unpin> TlsStream<S> {
+    async fn start_accept(mut self: Pin<&mut Self>) -> Result<bool> {
+        // TODO: suspend cert callback
+        let res = self.accept().await;
+
+        match res {
+            Ok(()) => Ok(true),
+            Err(e) => {
+                if e.etype == TLSWantX509Lookup {
+                    Ok(false)
+                } else {
+                    Err(e)
+                }
+            }
+        }
+    }
+
+    async fn resume_accept(mut self: Pin<&mut Self>) -> Result<()> {
+        // TODO: unblock cert callback
+        self.accept().await
+    }
+}
+
+async fn prepare_tls_stream<S: IO>(acceptor: &Acceptor, io: S) -> Result<TlsStream<S>> {
+    TlsStream::from_acceptor(acceptor, io)
+        .await
+        .explain_err(TLSHandshakeFailure, |e| format!("tls stream error: {e}"))
+}
+
+/// Perform TLS handshake for the given connection with the given configuration
+pub async fn handshake<S: IO>(acceptor: &Acceptor, io: S) -> Result<TlsStream<S>> {
+    let mut stream = prepare_tls_stream(acceptor, io).await?;
+    stream
+        .accept()
+        .await
+        .explain_err(TLSHandshakeFailure, |e| format!("TLS accept() failed: {e}"))?;
+    Ok(stream)
+}
+
+/// Perform TLS handshake for the given connection with the given configuration and callbacks
+/// callbacks are currently not supported within pingora Rustls and are ignored
+pub async fn handshake_with_callback<S: IO>(
+    acceptor: &Acceptor,
+    io: S,
+    _callbacks: &TlsAcceptCallbacks,
+) -> Result<TlsStream<S>> {
+    let mut tls_stream = prepare_tls_stream(acceptor, io).await?;
+    let done = Pin::new(&mut tls_stream).start_accept().await?;
+    if !done {
+        // TODO: verify if/how callback in handshake can be done using Rustls
+        warn!("Callacks are not supported with feature \"rustls\".");
+
+        Pin::new(&mut tls_stream)
+            .resume_accept()
+            .await
+            .explain_err(TLSHandshakeFailure, |e| format!("TLS accept() failed: {e}"))?;
+        Ok(tls_stream)
+    } else {
+        Ok(tls_stream)
+    }
+}
+
+#[async_trait]
+impl<S> Shutdown for TlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Sync + Unpin + Send,
+{
+    async fn shutdown(&mut self) {
+        match <Self as AsyncWriteExt>::shutdown(self).await {
+            Ok(()) => {}
+            Err(e) => {
+                warn!("TLS shutdown failed, {e}");
+            }
+        }
+    }
+}
+
+#[ignore]
+#[tokio::test]
+async fn test_async_cert() {
+    todo!("callback support and test for Rustls")
+}
diff --git a/pingora-core/src/protocols/tls/rustls/stream.rs b/pingora-core/src/protocols/tls/rustls/stream.rs
new file mode 100644
index 0000000..385a51e
--- /dev/null
+++ b/pingora-core/src/protocols/tls/rustls/stream.rs
@@ -0,0 +1,415 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io::{Error as IoError, ErrorKind as IoErrorKind, Result as IoResult};
+use std::ops::{Deref, DerefMut};
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::{Duration, SystemTime};
+
+use crate::listeners::tls::Acceptor;
+use crate::protocols::raw_connect::ProxyDigest;
+use crate::protocols::{tls::SslDigest, TimingDigest};
+use crate::protocols::{
+    GetProxyDigest, GetSocketDigest, GetTimingDigest, SocketDigest, Ssl, UniqueID, ALPN,
+};
+use crate::utils::tls::get_organization_serial;
+use pingora_error::ErrorType::{AcceptError, ConnectError, InternalError, TLSHandshakeFailure};
+use pingora_error::{Error, ImmutStr, OrErr, Result};
+use pingora_rustls::TlsStream as RusTlsStream;
+use pingora_rustls::{hash_certificate, NoDebug};
+use pingora_rustls::{Accept, Connect, ServerName, TlsConnector};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use x509_parser::nom::AsBytes;
+
+#[derive(Debug)]
+pub struct InnerStream<T> {
+    pub(crate) stream: Option<RusTlsStream<T>>,
+    connect: NoDebug<Option<Connect<T>>>,
+    accept: NoDebug<Option<Accept<T>>>,
+}
+
+/// The TLS connection
+#[derive(Debug)]
+pub struct TlsStream<T> {
+    tls: InnerStream<T>,
+    digest: Option<Arc<SslDigest>>,
+    timing: TimingDigest,
+}
+
+impl<T> TlsStream<T>
+where
+    T: AsyncRead + AsyncWrite + Unpin + Send,
+{
+    /// Create a new TLS connection from the given `stream`
+    ///
+    /// Using RustTLS the stream is only returned after the handshake.
+    /// The caller does therefor not need to perform [`Self::connect()`].
+    pub async fn from_connector(connector: &TlsConnector, domain: &str, stream: T) -> Result<Self> {
+        let server = ServerName::try_from(domain)
+            .map_err(|e| IoError::new(IoErrorKind::InvalidInput, e))
+            .explain_err(InternalError, |e| {
+                format!("failed to parse domain: {}, error: {}", domain, e)
+            })?
+            .to_owned();
+
+        let tls = InnerStream::from_connector(connector, server, stream)
+            .await
+            .explain_err(TLSHandshakeFailure, |e| format!("tls stream error: {e}"))?;
+
+        Ok(TlsStream {
+            tls,
+            digest: None,
+            timing: Default::default(),
+        })
+    }
+
+    /// Create a new TLS connection from the given `stream`
+    ///
+    /// Using RustTLS the stream is only returned after the handshake.
+    /// The caller does therefor not need to perform [`Self::accept()`].
+    pub(crate) async fn from_acceptor(acceptor: &Acceptor, stream: T) -> Result<Self> {
+        let tls = InnerStream::from_acceptor(acceptor, stream)
+            .await
+            .explain_err(TLSHandshakeFailure, |e| format!("tls stream error: {e}"))?;
+
+        Ok(TlsStream {
+            tls,
+            digest: None,
+            timing: Default::default(),
+        })
+    }
+}
+
+impl<S> GetSocketDigest for TlsStream<S>
+where
+    S: GetSocketDigest,
+{
+    fn get_socket_digest(&self) -> Option<Arc<SocketDigest>> {
+        self.tls.get_socket_digest()
+    }
+    fn set_socket_digest(&mut self, socket_digest: SocketDigest) {
+        self.tls.set_socket_digest(socket_digest)
+    }
+}
+
+impl<S> GetTimingDigest for TlsStream<S>
+where
+    S: GetTimingDigest,
+{
+    fn get_timing_digest(&self) -> Vec<Option<TimingDigest>> {
+        let mut ts_vec = self.tls.get_timing_digest();
+        ts_vec.push(Some(self.timing.clone()));
+        ts_vec
+    }
+    fn get_read_pending_time(&self) -> Duration {
+        self.tls.get_read_pending_time()
+    }
+
+    fn get_write_pending_time(&self) -> Duration {
+        self.tls.get_write_pending_time()
+    }
+}
+
+impl<S> GetProxyDigest for TlsStream<S>
+where
+    S: GetProxyDigest,
+{
+    fn get_proxy_digest(&self) -> Option<Arc<ProxyDigest>> {
+        self.tls.get_proxy_digest()
+    }
+}
+
+impl<T> TlsStream<T> {
+    pub fn ssl_digest(&self) -> Option<Arc<SslDigest>> {
+        self.digest.clone()
+    }
+}
+
+impl<T> Deref for TlsStream<T> {
+    type Target = InnerStream<T>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.tls
+    }
+}
+
+impl<T> DerefMut for TlsStream<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.tls
+    }
+}
+
+impl<T> TlsStream<T>
+where
+    T: AsyncRead + AsyncWrite + Unpin + Send,
+{
+    /// Connect to the remote TLS server as a client
+    pub(crate) async fn connect(&mut self) -> Result<()> {
+        self.tls.connect().await?;
+        self.timing.established_ts = SystemTime::now();
+        self.digest = self.tls.digest();
+        Ok(())
+    }
+
+    /// Finish the TLS handshake from client as a server
+    pub(crate) async fn accept(&mut self) -> Result<()> {
+        self.tls.accept().await?;
+        self.timing.established_ts = SystemTime::now();
+        self.digest = self.tls.digest();
+        Ok(())
+    }
+}
+
+impl<T> AsyncRead for TlsStream<T>
+where
+    T: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<IoResult<()>> {
+        Pin::new(&mut self.tls.stream.as_mut().unwrap()).poll_read(cx, buf)
+    }
+}
+
+impl<T> AsyncWrite for TlsStream<T>
+where
+    T: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<IoResult<usize>> {
+        Pin::new(&mut self.tls.stream.as_mut().unwrap()).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<IoResult<()>> {
+        Pin::new(&mut self.tls.stream.as_mut().unwrap()).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<IoResult<()>> {
+        Pin::new(&mut self.tls.stream.as_mut().unwrap()).poll_shutdown(cx)
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<IoResult<usize>> {
+        Pin::new(&mut self.tls.stream.as_mut().unwrap()).poll_write_vectored(cx, bufs)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        true
+    }
+}
+
+impl<T> UniqueID for TlsStream<T>
+where
+    T: UniqueID,
+{
+    fn id(&self) -> i32 {
+        self.tls.stream.as_ref().unwrap().get_ref().0.id()
+    }
+}
+
+impl<T> Ssl for TlsStream<T> {
+    fn get_ssl_digest(&self) -> Option<Arc<SslDigest>> {
+        self.ssl_digest()
+    }
+
+    fn selected_alpn_proto(&self) -> Option<ALPN> {
+        let st = self.tls.stream.as_ref();
+        if let Some(stream) = st {
+            let proto = stream.get_ref().1.alpn_protocol();
+            match proto {
+                None => None,
+                Some(raw) => ALPN::from_wire_selected(raw),
+            }
+        } else {
+            None
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + Unpin> InnerStream<T> {
+    /// Create a new TLS connection from the given `stream`
+    ///
+    /// The caller needs to perform [`Self::connect()`] or [`Self::accept()`] to perform TLS
+    /// handshake after.
+    pub(crate) async fn from_connector(
+        connector: &TlsConnector,
+        server: ServerName<'_>,
+        stream: T,
+    ) -> Result<Self> {
+        let connect = connector.connect(server.to_owned(), stream);
+        Ok(InnerStream {
+            accept: None.into(),
+            connect: Some(connect).into(),
+            stream: None,
+        })
+    }
+
+    pub(crate) async fn from_acceptor(acceptor: &Acceptor, stream: T) -> Result<Self> {
+        let accept = acceptor.acceptor.accept(stream);
+
+        Ok(InnerStream {
+            accept: Some(accept).into(),
+            connect: None.into(),
+            stream: None,
+        })
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + Unpin + Send> InnerStream<T> {
+    /// Connect to the remote TLS server as a client
+    pub(crate) async fn connect(&mut self) -> Result<()> {
+        let connect = &mut (*self.connect);
+
+        if let Some(ref mut connect) = connect {
+            let stream = connect
+                .await
+                .explain_err(TLSHandshakeFailure, |e| format!("tls connect error: {e}"))?;
+            self.stream = Some(RusTlsStream::Client(stream));
+            self.connect = None.into();
+
+            Ok(())
+        } else {
+            Err(Error::explain(
+                ConnectError,
+                ImmutStr::from("TLS connect not available to perform handshake."),
+            ))
+        }
+    }
+
+    /// Finish the TLS handshake from client as a server
+    /// no-op implementation within Rustls, handshake is performed during creation of stream.
+    pub(crate) async fn accept(&mut self) -> Result<()> {
+        let accept = &mut (*self.accept);
+
+        if let Some(ref mut accept) = accept {
+            let stream = accept
+                .await
+                .explain_err(TLSHandshakeFailure, |e| format!("tls connect error: {e}"))?;
+            self.stream = Some(RusTlsStream::Server(stream));
+            self.connect = None.into();
+
+            Ok(())
+        } else {
+            Err(Error::explain(
+                AcceptError,
+                ImmutStr::from("TLS accept not available to perform handshake."),
+            ))
+        }
+    }
+
+    pub(crate) fn digest(&mut self) -> Option<Arc<SslDigest>> {
+        Some(Arc::new(SslDigest::from_stream(&self.stream)))
+    }
+}
+
+impl<S> GetSocketDigest for InnerStream<S>
+where
+    S: GetSocketDigest,
+{
+    fn get_socket_digest(&self) -> Option<Arc<SocketDigest>> {
+        if let Some(stream) = self.stream.as_ref() {
+            stream.get_ref().0.get_socket_digest()
+        } else {
+            None
+        }
+    }
+    fn set_socket_digest(&mut self, socket_digest: SocketDigest) {
+        self.stream
+            .as_mut()
+            .unwrap()
+            .get_mut()
+            .0
+            .set_socket_digest(socket_digest)
+    }
+}
+
+impl<S> GetTimingDigest for InnerStream<S>
+where
+    S: GetTimingDigest,
+{
+    fn get_timing_digest(&self) -> Vec<Option<TimingDigest>> {
+        self.stream
+            .as_ref()
+            .unwrap()
+            .get_ref()
+            .0
+            .get_timing_digest()
+    }
+}
+
+impl<S> GetProxyDigest for InnerStream<S>
+where
+    S: GetProxyDigest,
+{
+    fn get_proxy_digest(&self) -> Option<Arc<ProxyDigest>> {
+        if let Some(stream) = self.stream.as_ref() {
+            stream.get_ref().0.get_proxy_digest()
+        } else {
+            None
+        }
+    }
+}
+
+impl SslDigest {
+    fn from_stream<T>(stream: &Option<RusTlsStream<T>>) -> Self {
+        let stream = stream.as_ref().unwrap();
+        let (_io, session) = stream.get_ref();
+        let protocol = session.protocol_version();
+        let cipher_suite = session.negotiated_cipher_suite();
+        let peer_certificates = session.peer_certificates();
+
+        let cipher = match cipher_suite {
+            Some(suite) => suite.suite().as_str().unwrap_or_default(),
+            None => "",
+        };
+
+        let version = match protocol {
+            Some(proto) => proto.as_str().unwrap_or_default(),
+            None => "",
+        };
+
+        let cert_digest = match peer_certificates {
+            Some(certs) => match certs.first() {
+                Some(cert) => hash_certificate(cert.clone()),
+                None => vec![],
+            },
+            None => vec![],
+        };
+
+        let (organization, serial_number) = match peer_certificates {
+            Some(certs) => match certs.first() {
+                Some(cert) => {
+                    let (organization, serial) = get_organization_serial(cert.as_bytes());
+                    (organization, Some(serial))
+                }
+                None => (None, None),
+            },
+            None => (None, None),
+        };
+
+        SslDigest {
+            cipher,
+            version,
+            organization,
+            serial_number,
+            cert_digest,
+        }
+    }
+}
diff --git a/pingora-core/src/utils/tls/mod.rs b/pingora-core/src/utils/tls/mod.rs
index 0bcaeac..4f59fcb 100644
--- a/pingora-core/src/utils/tls/mod.rs
+++ b/pingora-core/src/utils/tls/mod.rs
@@ -17,3 +17,9 @@ mod boringssl_openssl;
 
 #[cfg(feature = "openssl_derived")]
 pub use boringssl_openssl::*;
+
+#[cfg(feature = "rustls")]
+mod rustls;
+
+#[cfg(feature = "rustls")]
+pub use rustls::*;
diff --git a/pingora-core/src/utils/tls/rustls.rs b/pingora-core/src/utils/tls/rustls.rs
new file mode 100644
index 0000000..44c3999
--- /dev/null
+++ b/pingora-core/src/utils/tls/rustls.rs
@@ -0,0 +1,157 @@
+// Copyright 2024 Cloudflare, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::protocols::tls::CertWrapper;
+use pingora_error::Result;
+use std::hash::{Hash, Hasher};
+use x509_parser::prelude::FromDer;
+
+pub fn get_organization_serial(cert: &[u8]) -> (Option<String>, String) {
+    let serial = get_serial(cert).expect("Failed to get serial for certificate.");
+    (get_organization(cert), serial)
+}
+
+pub fn get_serial(cert: &[u8]) -> Result<String> {
+    let (_, x509cert) = x509_parser::certificate::X509Certificate::from_der(cert)
+        .expect("Failed to parse certificate from DER format.");
+    Ok(x509cert.raw_serial_as_string())
+}
+
+/// Return the organization associated with the X509 certificate.
+pub fn get_organization(cert: &[u8]) -> Option<String> {
+    let (_, x509cert) = x509_parser::certificate::X509Certificate::from_der(cert)
+        .expect("Failed to parse certificate from DER format.");
+    x509cert
+        .subject
+        .iter_organization()
+        .filter_map(|a| a.as_str().ok())
+        .map(|a| a.to_string())
+        .reduce(|cur, next| cur + &next)
+}
+
+/// Return the organization unit associated with the X509 certificate.
+pub fn get_organization_unit(cert: &CertWrapper) -> Option<String> {
+    get_organization_unit_bytes(&cert.0)
+}
+
+/// Return the organization unit associated with the X509 certificate.
+pub fn get_organization_unit_bytes(cert: &[u8]) -> Option<String> {
+    let (_, x509cert) = x509_parser::certificate::X509Certificate::from_der(cert)
+        .expect("Failed to parse certificate from DER format.");
+    x509cert
+        .subject
+        .iter_organizational_unit()
+        .filter_map(|a| a.as_str().ok())
+        .map(|a| a.to_string())
+        .reduce(|cur, next| cur + &next)
+}
+
+pub fn get_common_name(cert: &[u8]) -> Option<String> {
+    let (_, x509cert) = x509_parser::certificate::X509Certificate::from_der(cert)
+        .expect("Failed to parse certificate from DER format.");
+    x509cert
+        .subject
+        .iter_common_name()
+        .filter_map(|a| a.as_str().ok())
+        .map(|a| a.to_string())
+        .reduce(|cur, next| cur + &next)
+}
+
+/// Return the organization unit associated with the X509 certificate.
+pub fn get_not_after(cert: &[u8]) -> String {
+    let (_, x509cert) = x509_parser::certificate::X509Certificate::from_der(cert)
+        .expect("Failed to parse certificate from DER format.");
+    x509cert.validity.not_after.to_string()
+}
+
+/// This type contains a list of one or more certificates and an associated private key. The leaf
+/// certificate should always be first. The certificates and keys are stored in Vec<u8> DER encoded
+/// form for usage within OpenSSL/BoringSSL & RusTLS.
+#[derive(Clone)]
+pub struct CertKey {
+    certificates: Vec<Vec<u8>>,
+    key: Vec<u8>,
+}
+
+impl CertKey {
+    /// Create a new `CertKey` given a list of certificates and a private key.
+    pub fn new(certificates: Vec<Vec<u8>>, key: Vec<u8>) -> CertKey {
+        assert!(
+            !certificates.is_empty() && !certificates.first().unwrap().is_empty(),
+            "expected a non-empty vector of certificates in CertKey::new"
+        );
+
+        CertKey { certificates, key }
+    }
+
+    /// Peek at the leaf certificate.
+    pub fn leaf(&self) -> &Vec<u8> {
+        // This is safe due to the assertion above.
+        &self.certificates[0]
+    }
+
+    /// Return the key.
+    pub fn key(&self) -> &Vec<u8> {
+        &self.key
+    }
+
+    /// Return a slice of intermediate certificates. An empty slice means there are none.
+    pub fn intermediates(&self) -> Vec<&Vec<u8>> {
+        self.certificates.iter().skip(1).collect()
+    }
+
+    /// Return the organization from the leaf certificate.
+    pub fn organization(&self) -> Option<String> {
+        get_organization(self.leaf())
+    }
+
+    /// Return the serial from the leaf certificate.
+    pub fn serial(&self) -> String {
+        get_serial(self.leaf()).unwrap()
+    }
+}
+
+// hide private key
+impl std::fmt::Debug for CertKey {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("CertKey")
+            .field("X509", &self.leaf())
+            .finish()
+    }
+}
+
+impl std::fmt::Display for CertKey {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let leaf = self.leaf();
+        if let Some(cn) = get_common_name(leaf) {
+            // Write CN if it exists
+            write!(f, "CN: {cn},")?;
+        } else if let Some(org_unit) = get_organization_unit_bytes(leaf) {
+            // CA cert might not have CN, so print its unit name instead
+            write!(f, "Org Unit: {org_unit},")?;
+        }
+        write!(f, ", expire: {}", get_not_after(leaf))
+        // ignore the details of the private key
+    }
+}
+
+impl Hash for CertKey {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        for certificate in &self.certificates {
+            if let Ok(serial) = get_serial(certificate) {
+                serial.hash(state)
+            }
+        }
+    }
+}
diff --git a/pingora-proxy/tests/keys/server.crt b/pingora-proxy/tests/keys/server.crt
index afb2d1e..28cdadf 100644
--- a/pingora-proxy/tests/keys/server.crt
+++ b/pingora-proxy/tests/keys/server.crt
@@ -1,13 +1,14 @@
 -----BEGIN CERTIFICATE-----
-MIIB9zCCAZ2gAwIBAgIUMI7aLvTxyRFCHhw57hGt4U6yupcwCgYIKoZIzj0EAwIw
+MIICJzCCAc6gAwIBAgIUU+G0acG/uiMu1ZDSjlcoY4gH53QwCgYIKoZIzj0EAwIw
 ZDELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMRYwFAYDVQQHDA1TYW4gRnJhbmNp
 c2NvMRgwFgYDVQQKDA9DbG91ZGZsYXJlLCBJbmMxFjAUBgNVBAMMDW9wZW5ydXN0
-eS5vcmcwHhcNMjIwNDExMjExMzEzWhcNMzIwNDA4MjExMzEzWjBkMQswCQYDVQQG
+eS5vcmcwHhcNMjQwNzI0MTMzOTQ4WhcNMzQwNzIyMTMzOTQ4WjBkMQswCQYDVQQG
 EwJVUzELMAkGA1UECAwCQ0ExFjAUBgNVBAcMDVNhbiBGcmFuY2lzY28xGDAWBgNV
 BAoMD0Nsb3VkZmxhcmUsIEluYzEWMBQGA1UEAwwNb3BlbnJ1c3R5Lm9yZzBZMBMG
 ByqGSM49AgEGCCqGSM49AwEHA0IABNn/9RZtR48knaJD6tk9BdccaJfZ0hGEPn6B
-SDXmlmJPhcTBqa4iUwW/ABpGvO3FpJcNWasrX2k+qZLq3g205MKjLTArMCkGA1Ud
-EQQiMCCCDyoub3BlbnJ1c3R5Lm9yZ4INb3BlbnJ1c3R5Lm9yZzAKBggqhkjOPQQD
-AgNIADBFAiAjISZ9aEKmobKGlT76idO740J6jPaX/hOrm41MLeg69AIhAJqKrSyz
-wD/AAF5fR6tXmBqlnpQOmtxfdy13wDr4MT3h
------END CERTIFICATE-----
+SDXmlmJPhcTBqa4iUwW/ABpGvO3FpJcNWasrX2k+qZLq3g205MKjXjBcMDsGA1Ud
+EQQ0MDKCDyoub3BlbnJ1c3R5Lm9yZ4INb3BlbnJ1c3R5Lm9yZ4IHY2F0LmNvbYIH
+ZG9nLmNvbTAdBgNVHQ4EFgQUnfYAFWyQnSN57IGokj7jcz8ChJQwCgYIKoZIzj0E
+AwIDRwAwRAIgQr+Ly2cH04CncbnbhUf4hBl5frTp1pXgGnn8dYjd+UcCICuunEtp
+H/a42/sVGBFvjS6FOFe6ZDs4oWBNEqQSw0S2
+-----END CERTIFICATE-----
\ No newline at end of file
diff --git a/pingora-proxy/tests/utils/conf/keys/server.crt b/pingora-proxy/tests/utils/conf/keys/server.crt
index afb2d1e..28cdadf 100644
--- a/pingora-proxy/tests/utils/conf/keys/server.crt
+++ b/pingora-proxy/tests/utils/conf/keys/server.crt
@@ -1,13 +1,14 @@
 -----BEGIN CERTIFICATE-----
-MIIB9zCCAZ2gAwIBAgIUMI7aLvTxyRFCHhw57hGt4U6yupcwCgYIKoZIzj0EAwIw
+MIICJzCCAc6gAwIBAgIUU+G0acG/uiMu1ZDSjlcoY4gH53QwCgYIKoZIzj0EAwIw
 ZDELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMRYwFAYDVQQHDA1TYW4gRnJhbmNp
 c2NvMRgwFgYDVQQKDA9DbG91ZGZsYXJlLCBJbmMxFjAUBgNVBAMMDW9wZW5ydXN0
-eS5vcmcwHhcNMjIwNDExMjExMzEzWhcNMzIwNDA4MjExMzEzWjBkMQswCQYDVQQG
+eS5vcmcwHhcNMjQwNzI0MTMzOTQ4WhcNMzQwNzIyMTMzOTQ4WjBkMQswCQYDVQQG
 EwJVUzELMAkGA1UECAwCQ0ExFjAUBgNVBAcMDVNhbiBGcmFuY2lzY28xGDAWBgNV
 BAoMD0Nsb3VkZmxhcmUsIEluYzEWMBQGA1UEAwwNb3BlbnJ1c3R5Lm9yZzBZMBMG
 ByqGSM49AgEGCCqGSM49AwEHA0IABNn/9RZtR48knaJD6tk9BdccaJfZ0hGEPn6B
-SDXmlmJPhcTBqa4iUwW/ABpGvO3FpJcNWasrX2k+qZLq3g205MKjLTArMCkGA1Ud
-EQQiMCCCDyoub3BlbnJ1c3R5Lm9yZ4INb3BlbnJ1c3R5Lm9yZzAKBggqhkjOPQQD
-AgNIADBFAiAjISZ9aEKmobKGlT76idO740J6jPaX/hOrm41MLeg69AIhAJqKrSyz
-wD/AAF5fR6tXmBqlnpQOmtxfdy13wDr4MT3h
------END CERTIFICATE-----
+SDXmlmJPhcTBqa4iUwW/ABpGvO3FpJcNWasrX2k+qZLq3g205MKjXjBcMDsGA1Ud
+EQQ0MDKCDyoub3BlbnJ1c3R5Lm9yZ4INb3BlbnJ1c3R5Lm9yZ4IHY2F0LmNvbYIH
+ZG9nLmNvbTAdBgNVHQ4EFgQUnfYAFWyQnSN57IGokj7jcz8ChJQwCgYIKoZIzj0E
+AwIDRwAwRAIgQr+Ly2cH04CncbnbhUf4hBl5frTp1pXgGnn8dYjd+UcCICuunEtp
+H/a42/sVGBFvjS6FOFe6ZDs4oWBNEqQSw0S2
+-----END CERTIFICATE-----
\ No newline at end of file
diff --git a/pingora-rustls/Cargo.toml b/pingora-rustls/Cargo.toml
index 677cf18..1c84d30 100644
--- a/pingora-rustls/Cargo.toml
+++ b/pingora-rustls/Cargo.toml
@@ -13,3 +13,17 @@ RusTLS async APIs for Pingora.
 [lib]
 name = "pingora_rustls"
 path = "src/lib.rs"
+
+[dependencies]
+log = "0.4.21"
+ring = "0.17.8"
+rustls = "0.23.12"
+rustls-native-certs = "0.7.1"
+rustls-pemfile = "2.1.2"
+rustls-pki-types = "1.7.0"
+tokio-rustls = "0.26.0"
+no_debug = "3.1.0"
+
+[dev-dependencies]
+tokio-test = "0.4.3"
+tokio = { workspace = true, features = ["full"] }
diff --git a/pingora-rustls/src/lib.rs b/pingora-rustls/src/lib.rs
index 814bcb9..e4d5de0 100644
--- a/pingora-rustls/src/lib.rs
+++ b/pingora-rustls/src/lib.rs
@@ -12,6 +12,147 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub fn rustls() {
-    todo!()
+#![warn(clippy::all)]
+
+use std::fs::File;
+use std::io::BufReader;
+
+use log::{error, warn};
+pub use no_debug::{Ellipses, NoDebug, WithTypeInfo};
+pub use rustls::{version, ClientConfig, RootCertStore, ServerConfig, Stream};
+pub use rustls_native_certs::load_native_certs;
+use rustls_pemfile::Item;
+pub use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName};
+pub use tokio_rustls::client::TlsStream as ClientTlsStream;
+pub use tokio_rustls::server::TlsStream as ServerTlsStream;
+pub use tokio_rustls::{Accept, Connect, TlsAcceptor, TlsConnector, TlsStream};
+
+fn load_file(path: &String) -> BufReader<File> {
+    let file = File::open(path).expect("io error");
+    BufReader::new(file)
+}
+fn load_pem_file(path: &String) -> Result<Vec<Item>, std::io::Error> {
+    let iter: Vec<Item> = rustls_pemfile::read_all(&mut load_file(path))
+        .filter_map(|f| {
+            if let Ok(f) = f {
+                Some(f)
+            } else {
+                let err = f.err().unwrap();
+                warn!(
+                    "Skipping PEM element in file \"{}\" due to error \"{}\"",
+                    path, err
+                );
+                None
+            }
+        })
+        .collect();
+    Ok(iter)
+}
+
+pub fn load_ca_file_into_store(path: &String, cert_store: &mut RootCertStore) {
+    let ca_file = load_pem_file(path);
+    match ca_file {
+        Ok(cas) => {
+            cas.into_iter().for_each(|pem_item| {
+                // only loading certificates, handling a CA file
+                match pem_item {
+                    Item::X509Certificate(content) => match cert_store.add(content) {
+                        Ok(_) => {}
+                        Err(err) => {
+                            error!("{}", err)
+                        }
+                    },
+                    Item::Pkcs1Key(_) => {}
+                    Item::Pkcs8Key(_) => {}
+                    Item::Sec1Key(_) => {}
+                    Item::Crl(_) => {}
+                    Item::Csr(_) => {}
+                    _ => {}
+                }
+            });
+        }
+        Err(err) => {
+            error!(
+                "Failed to load configured ca file located at \"{}\", error: \"{}\"",
+                path, err
+            );
+        }
+    }
+}
+
+pub fn load_platform_certs_incl_env_into_store(ca_certs: &mut RootCertStore) {
+    // this includes handling of ENV vars SSL_CERT_FILE & SSL_CERT_DIR
+    let native_platform_certs = load_native_certs();
+    match native_platform_certs {
+        Ok(certs) => {
+            for cert in certs {
+                ca_certs.add(cert).unwrap();
+            }
+        }
+        Err(err) => {
+            error!(
+                "Failed to load native platform ca-certificates: \"{:?}\". Continuing without ...",
+                err
+            );
+        }
+    }
+}
+
+pub fn load_certs_key_file<'a>(
+    cert: &String,
+    key: &String,
+) -> Option<(Vec<CertificateDer<'a>>, PrivateKeyDer<'a>)> {
+    let certs_file = load_pem_file(cert)
+        .unwrap_or_else(|_| panic!("Failed to load configured cert file located at {}.", cert));
+    let key_file = load_pem_file(key)
+        .unwrap_or_else(|_| panic!("Failed to load configured key file located at {}.", cert));
+
+    let mut certs: Vec<CertificateDer<'a>> = vec![];
+    certs_file.into_iter().for_each(|i| {
+        if let Item::X509Certificate(cert) = i {
+            certs.push(cert)
+        }
+    });
+
+    let private_key = match key_file.into_iter().next()? {
+        Item::Pkcs1Key(key) => Some(PrivateKeyDer::from(key)),
+        Item::Pkcs8Key(key) => Some(PrivateKeyDer::from(key)),
+        Item::Sec1Key(key) => Some(PrivateKeyDer::from(key)),
+        _ => None,
+    };
+
+    if certs.is_empty() || private_key.is_none() {
+        None
+    } else {
+        Some((certs, private_key?))
+    }
+}
+
+pub fn load_pem_file_ca(path: &String) -> Vec<u8> {
+    let mut reader = load_file(path);
+    let cas_file = rustls_pemfile::certs(&mut reader);
+    let ca = cas_file.into_iter().find_map(|pem_item| {
+        if let Ok(item) = pem_item {
+            Some(item)
+        } else {
+            None
+        }
+    });
+    match ca {
+        None => Vec::new(),
+        Some(ca) => ca.to_vec(),
+    }
+}
+
+pub fn load_pem_file_private_key(path: &String) -> Vec<u8> {
+    let key = rustls_pemfile::private_key(&mut load_file(path));
+    if let Ok(Some(key)) = key {
+        return key.secret_der().to_vec();
+    }
+    Vec::new()
+}
+
+pub fn hash_certificate(cert: CertificateDer) -> Vec<u8> {
+    let hash = ring::digest::digest(&ring::digest::SHA256, cert.as_ref());
+    hash.as_ref().to_vec()
 }