#![warn(missing_docs)]
#![cfg_attr(docsrs, feature(doc_auto_cfg))]
use crate::cache::{SessionCache, SessionKey};
use antidote::Mutex;
use boring::error::ErrorStack;
use boring::ex_data::Index;
use boring::ssl::{
ConnectConfiguration, Ssl, SslConnector, SslConnectorBuilder, SslMethod, SslRef,
SslSessionCacheMode,
};
use http::uri::Scheme;
use hyper::client::connect::{Connected, Connection};
#[cfg(feature = "runtime")]
use hyper::client::HttpConnector;
use hyper::service::Service;
use hyper::Uri;
use once_cell::sync::OnceCell;
use std::fmt::Debug;
use std::future::Future;
use std::io;
use std::net;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::{error::Error, fmt};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio_boring::SslStream;
use tower_layer::Layer;
mod cache;
#[cfg(test)]
mod test;
fn key_index() -> Result<Index<Ssl, SessionKey>, ErrorStack> {
static IDX: OnceCell<Index<Ssl, SessionKey>> = OnceCell::new();
IDX.get_or_try_init(Ssl::new_ex_index).copied()
}
#[derive(Clone)]
struct Inner {
ssl: SslConnector,
cache: Arc<Mutex<SessionCache>>,
callback: Option<Callback>,
ssl_callback: Option<SslCallback>,
}
type Callback =
Arc<dyn Fn(&mut ConnectConfiguration, &Uri) -> Result<(), ErrorStack> + Sync + Send>;
type SslCallback = Arc<dyn Fn(&mut SslRef, &Uri) -> Result<(), ErrorStack> + Sync + Send>;
impl Inner {
fn setup_ssl(&self, uri: &Uri, host: &str) -> Result<Ssl, ErrorStack> {
let mut conf = self.ssl.configure()?;
if let Some(ref callback) = self.callback {
callback(&mut conf, uri)?;
}
let key = SessionKey {
host: host.to_string(),
port: uri.port_u16().unwrap_or(443),
};
if let Some(session) = self.cache.lock().get(&key) {
unsafe {
conf.set_session(&session)?;
}
}
let idx = key_index()?;
conf.set_ex_data(idx, key);
let mut ssl = conf.into_ssl(host)?;
if let Some(ref ssl_callback) = self.ssl_callback {
ssl_callback(&mut ssl, uri)?;
}
Ok(ssl)
}
}
pub struct HttpsLayer {
inner: Inner,
}
pub struct HttpsLayerSettings {
session_cache_capacity: usize,
}
impl HttpsLayerSettings {
pub fn builder() -> HttpsLayerSettingsBuilder {
HttpsLayerSettingsBuilder(HttpsLayerSettings::default())
}
}
impl Default for HttpsLayerSettings {
fn default() -> Self {
Self {
session_cache_capacity: 8,
}
}
}
pub struct HttpsLayerSettingsBuilder(HttpsLayerSettings);
impl HttpsLayerSettingsBuilder {
pub fn set_session_cache_capacity(&mut self, capacity: usize) {
self.0.session_cache_capacity = capacity;
}
pub fn build(self) -> HttpsLayerSettings {
self.0
}
}
impl HttpsLayer {
pub fn new() -> Result<HttpsLayer, ErrorStack> {
let mut ssl = SslConnector::builder(SslMethod::tls())?;
ssl.set_alpn_protos(b"\x02h2\x08http/1.1")?;
Self::with_connector(ssl)
}
pub fn with_connector(ssl: SslConnectorBuilder) -> Result<HttpsLayer, ErrorStack> {
Self::with_connector_and_settings(ssl, Default::default())
}
pub fn with_connector_and_settings(
mut ssl: SslConnectorBuilder,
settings: HttpsLayerSettings,
) -> Result<HttpsLayer, ErrorStack> {
let cache = Arc::new(Mutex::new(SessionCache::with_capacity(
settings.session_cache_capacity,
)));
ssl.set_session_cache_mode(SslSessionCacheMode::CLIENT);
ssl.set_new_session_callback({
let cache = cache.clone();
move |ssl, session| {
if let Some(key) = key_index().ok().and_then(|idx| ssl.ex_data(idx)) {
cache.lock().insert(key.clone(), session);
}
}
});
Ok(HttpsLayer {
inner: Inner {
ssl: ssl.build(),
cache,
callback: None,
ssl_callback: None,
},
})
}
pub fn set_callback<F>(&mut self, callback: F)
where
F: Fn(&mut ConnectConfiguration, &Uri) -> Result<(), ErrorStack> + 'static + Sync + Send,
{
self.inner.callback = Some(Arc::new(callback));
}
pub fn set_ssl_callback<F>(&mut self, callback: F)
where
F: Fn(&mut SslRef, &Uri) -> Result<(), ErrorStack> + 'static + Sync + Send,
{
self.inner.ssl_callback = Some(Arc::new(callback));
}
}
impl<S> Layer<S> for HttpsLayer {
type Service = HttpsConnector<S>;
fn layer(&self, inner: S) -> HttpsConnector<S> {
HttpsConnector {
http: inner,
inner: self.inner.clone(),
}
}
}
#[derive(Clone)]
pub struct HttpsConnector<T> {
http: T,
inner: Inner,
}
#[cfg(feature = "runtime")]
impl HttpsConnector<HttpConnector> {
pub fn new() -> Result<HttpsConnector<HttpConnector>, ErrorStack> {
let mut http = HttpConnector::new();
http.enforce_http(false);
HttpsLayer::new().map(|l| l.layer(http))
}
}
impl<S, T> HttpsConnector<S>
where
S: Service<Uri, Response = T> + Send,
S::Error: Into<Box<dyn Error + Send + Sync>>,
S::Future: Unpin + Send + 'static,
T: AsyncRead + AsyncWrite + Connection + Unpin + Debug + Sync + Send + 'static,
{
pub fn with_connector(
http: S,
ssl: SslConnectorBuilder,
) -> Result<HttpsConnector<S>, ErrorStack> {
HttpsLayer::with_connector(ssl).map(|l| l.layer(http))
}
pub fn set_callback<F>(&mut self, callback: F)
where
F: Fn(&mut ConnectConfiguration, &Uri) -> Result<(), ErrorStack> + 'static + Sync + Send,
{
self.inner.callback = Some(Arc::new(callback));
}
pub fn set_ssl_callback<F>(&mut self, callback: F)
where
F: Fn(&mut SslRef, &Uri) -> Result<(), ErrorStack> + 'static + Sync + Send,
{
self.inner.ssl_callback = Some(Arc::new(callback));
}
}
impl<S> Service<Uri> for HttpsConnector<S>
where
S: Service<Uri> + Send,
S::Error: Into<Box<dyn Error + Send + Sync>>,
S::Future: Unpin + Send + 'static,
S::Response: AsyncRead + AsyncWrite + Connection + Unpin + Debug + Sync + Send + 'static,
{
type Response = MaybeHttpsStream<S::Response>;
type Error = Box<dyn Error + Sync + Send>;
#[allow(clippy::type_complexity)]
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.http.poll_ready(cx).map_err(Into::into)
}
fn call(&mut self, uri: Uri) -> Self::Future {
let is_tls_scheme = uri
.scheme()
.map(|s| s == &Scheme::HTTPS || s.as_str() == "wss")
.unwrap_or(false);
let tls_setup = if is_tls_scheme {
Some((self.inner.clone(), uri.clone()))
} else {
None
};
let connect = self.http.call(uri);
let f = async {
let conn = connect.await.map_err(Into::into)?;
let (inner, uri) = match tls_setup {
Some((inner, uri)) => (inner, uri),
None => return Ok(MaybeHttpsStream::Http(conn)),
};
let mut host = uri.host().ok_or("URI missing host")?;
if !host.is_empty() {
let last = host.len() - 1;
let mut chars = host.chars();
if let (Some('['), Some(']')) = (chars.next(), chars.last()) {
if host[1..last].parse::<net::Ipv6Addr>().is_ok() {
host = &host[1..last];
}
}
}
let ssl = inner.setup_ssl(&uri, host)?;
let stream = tokio_boring::SslStreamBuilder::new(ssl, conn)
.connect()
.await?;
Ok(MaybeHttpsStream::Https(stream))
};
Box::pin(f)
}
}
pub enum MaybeHttpsStream<T> {
Http(T),
Https(SslStream<T>),
}
impl<T> AsyncRead for MaybeHttpsStream<T>
where
T: AsyncRead + AsyncWrite + Unpin,
{
fn poll_read(
mut self: Pin<&mut Self>,
ctx: &mut Context<'_>,
buf: &mut ReadBuf,
) -> Poll<io::Result<()>> {
match &mut *self {
MaybeHttpsStream::Http(s) => Pin::new(s).poll_read(ctx, buf),
MaybeHttpsStream::Https(s) => Pin::new(s).poll_read(ctx, buf),
}
}
}
impl<T> AsyncWrite for MaybeHttpsStream<T>
where
T: AsyncRead + AsyncWrite + Unpin,
{
fn poll_write(
mut self: Pin<&mut Self>,
ctx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
match &mut *self {
MaybeHttpsStream::Http(s) => Pin::new(s).poll_write(ctx, buf),
MaybeHttpsStream::Https(s) => Pin::new(s).poll_write(ctx, buf),
}
}
fn poll_flush(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<io::Result<()>> {
match &mut *self {
MaybeHttpsStream::Http(s) => Pin::new(s).poll_flush(ctx),
MaybeHttpsStream::Https(s) => Pin::new(s).poll_flush(ctx),
}
}
fn poll_shutdown(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<io::Result<()>> {
match &mut *self {
MaybeHttpsStream::Http(s) => Pin::new(s).poll_shutdown(ctx),
MaybeHttpsStream::Https(s) => Pin::new(s).poll_shutdown(ctx),
}
}
}
impl<T> Connection for MaybeHttpsStream<T>
where
T: Connection,
{
fn connected(&self) -> Connected {
match self {
MaybeHttpsStream::Http(s) => s.connected(),
MaybeHttpsStream::Https(s) => {
let mut connected = s.get_ref().connected();
if s.ssl().selected_alpn_protocol() == Some(b"h2") {
connected = connected.negotiated_h2();
}
connected
}
}
}
}
impl<T> fmt::Debug for MaybeHttpsStream<T> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match *self {
MaybeHttpsStream::Http(..) => f.pad("Http(..)"),
MaybeHttpsStream::Https(..) => f.pad("Https(..)"),
}
}
}