2021-12-09 14:43:19 +00:00
|
|
|
use std::convert::Infallible;
|
2021-12-07 14:20:45 +00:00
|
|
|
use std::net::SocketAddr;
|
2021-12-07 15:37:22 +00:00
|
|
|
use std::sync::{atomic::Ordering, Arc};
|
2021-12-07 14:20:45 +00:00
|
|
|
|
|
|
|
|
use anyhow::Result;
|
|
|
|
|
use log::*;
|
|
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
use accept_encoding_fork::Encoding;
|
|
|
|
|
use async_compression::tokio::bufread::*;
|
|
|
|
|
use futures::TryStreamExt;
|
2021-12-07 17:50:58 +00:00
|
|
|
use http::header::{HeaderName, HeaderValue};
|
2021-12-07 14:20:45 +00:00
|
|
|
use hyper::server::conn::Http;
|
|
|
|
|
use hyper::service::service_fn;
|
2021-12-09 14:43:19 +00:00
|
|
|
use hyper::{header, Body, Request, Response, StatusCode};
|
2021-12-07 14:20:45 +00:00
|
|
|
use tokio::net::TcpListener;
|
|
|
|
|
use tokio::sync::watch;
|
|
|
|
|
use tokio_rustls::TlsAcceptor;
|
2021-12-09 14:43:19 +00:00
|
|
|
use tokio_util::io::{ReaderStream, StreamReader};
|
2021-12-07 14:20:45 +00:00
|
|
|
|
|
|
|
|
use crate::cert_store::{CertStore, StoreResolver};
|
|
|
|
|
use crate::proxy_config::ProxyConfig;
|
|
|
|
|
use crate::reverse_proxy;
|
|
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
pub struct HttpsConfig {
|
|
|
|
|
pub bind_addr: SocketAddr,
|
|
|
|
|
pub enable_compression: bool,
|
|
|
|
|
pub compress_mime_types: Vec<String>,
|
|
|
|
|
}
|
|
|
|
|
|
2021-12-07 14:20:45 +00:00
|
|
|
pub async fn serve_https(
|
2021-12-09 14:43:19 +00:00
|
|
|
config: HttpsConfig,
|
2021-12-07 14:20:45 +00:00
|
|
|
cert_store: Arc<CertStore>,
|
2021-12-09 14:43:19 +00:00
|
|
|
rx_proxy_config: watch::Receiver<Arc<ProxyConfig>>,
|
2021-12-07 14:20:45 +00:00
|
|
|
) -> Result<()> {
|
2021-12-09 14:43:19 +00:00
|
|
|
let config = Arc::new(config);
|
|
|
|
|
|
|
|
|
|
let mut tls_cfg = rustls::ServerConfig::builder()
|
2021-12-07 14:20:45 +00:00
|
|
|
.with_safe_defaults()
|
|
|
|
|
.with_no_client_auth()
|
|
|
|
|
.with_cert_resolver(Arc::new(StoreResolver(cert_store)));
|
|
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
tls_cfg.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
|
|
|
|
|
let tls_acceptor = Arc::new(TlsAcceptor::from(Arc::new(tls_cfg)));
|
2021-12-07 14:20:45 +00:00
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
info!("Starting to serve on https://{}.", config.bind_addr);
|
2021-12-07 14:20:45 +00:00
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
let tcp = TcpListener::bind(config.bind_addr).await?;
|
2021-12-07 14:20:45 +00:00
|
|
|
loop {
|
|
|
|
|
let (socket, remote_addr) = tcp.accept().await?;
|
|
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
let rx_proxy_config = rx_proxy_config.clone();
|
2021-12-07 14:20:45 +00:00
|
|
|
let tls_acceptor = tls_acceptor.clone();
|
2021-12-09 14:43:19 +00:00
|
|
|
let config = config.clone();
|
2021-12-07 14:20:45 +00:00
|
|
|
|
|
|
|
|
tokio::spawn(async move {
|
|
|
|
|
match tls_acceptor.accept(socket).await {
|
|
|
|
|
Ok(stream) => {
|
|
|
|
|
debug!("TLS handshake was successfull");
|
|
|
|
|
let http_result = Http::new()
|
|
|
|
|
.serve_connection(
|
|
|
|
|
stream,
|
|
|
|
|
service_fn(move |req: Request<Body>| {
|
2021-12-09 14:43:19 +00:00
|
|
|
let https_config = config.clone();
|
|
|
|
|
let proxy_config: Arc<ProxyConfig> =
|
|
|
|
|
rx_proxy_config.borrow().clone();
|
|
|
|
|
handle_outer(remote_addr, req, https_config, proxy_config)
|
2021-12-07 14:20:45 +00:00
|
|
|
}),
|
|
|
|
|
)
|
|
|
|
|
.await;
|
|
|
|
|
if let Err(http_err) = http_result {
|
2021-12-08 10:24:25 +00:00
|
|
|
warn!("HTTP error: {}", http_err);
|
2021-12-07 14:20:45 +00:00
|
|
|
}
|
|
|
|
|
}
|
2021-12-08 10:24:25 +00:00
|
|
|
Err(e) => warn!("Error in TLS connection: {}", e),
|
2021-12-07 14:20:45 +00:00
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
async fn handle_outer(
|
|
|
|
|
remote_addr: SocketAddr,
|
|
|
|
|
req: Request<Body>,
|
|
|
|
|
https_config: Arc<HttpsConfig>,
|
|
|
|
|
proxy_config: Arc<ProxyConfig>,
|
|
|
|
|
) -> Result<Response<Body>, Infallible> {
|
|
|
|
|
match handle(remote_addr, req, https_config, proxy_config).await {
|
|
|
|
|
Err(e) => {
|
|
|
|
|
warn!("Handler error: {}", e);
|
|
|
|
|
Ok(Response::builder()
|
|
|
|
|
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
|
|
|
|
.body(Body::from(format!("{}", e)))
|
|
|
|
|
.unwrap())
|
|
|
|
|
}
|
|
|
|
|
Ok(r) => Ok(r),
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2021-12-07 14:20:45 +00:00
|
|
|
// Custom echo service, handling two different routes and a
|
|
|
|
|
// catch-all 404 responder.
|
|
|
|
|
async fn handle(
|
|
|
|
|
remote_addr: SocketAddr,
|
|
|
|
|
req: Request<Body>,
|
2021-12-09 14:43:19 +00:00
|
|
|
https_config: Arc<HttpsConfig>,
|
2021-12-07 14:20:45 +00:00
|
|
|
proxy_config: Arc<ProxyConfig>,
|
|
|
|
|
) -> Result<Response<Body>, anyhow::Error> {
|
2021-12-08 21:58:19 +00:00
|
|
|
let method = req.method().clone();
|
|
|
|
|
let uri = req.uri().to_string();
|
|
|
|
|
|
2021-12-07 14:20:45 +00:00
|
|
|
let host = if let Some(auth) = req.uri().authority() {
|
|
|
|
|
auth.as_str()
|
|
|
|
|
} else {
|
|
|
|
|
req.headers()
|
|
|
|
|
.get("host")
|
|
|
|
|
.ok_or_else(|| anyhow!("Missing host header"))?
|
|
|
|
|
.to_str()?
|
|
|
|
|
};
|
|
|
|
|
let path = req.uri().path();
|
2021-12-09 15:03:19 +00:00
|
|
|
let accept_encoding = accept_encoding_fork::encodings(req.headers()).unwrap_or(vec![]);
|
2021-12-07 14:20:45 +00:00
|
|
|
|
2021-12-08 21:58:19 +00:00
|
|
|
let best_match = proxy_config
|
2021-12-07 14:20:45 +00:00
|
|
|
.entries
|
|
|
|
|
.iter()
|
|
|
|
|
.filter(|ent| {
|
2021-12-08 10:11:22 +00:00
|
|
|
ent.host.matches(host)
|
2021-12-07 14:20:45 +00:00
|
|
|
&& ent
|
|
|
|
|
.path_prefix
|
|
|
|
|
.as_ref()
|
|
|
|
|
.map(|prefix| path.starts_with(prefix))
|
|
|
|
|
.unwrap_or(true)
|
|
|
|
|
})
|
2021-12-08 21:29:08 +00:00
|
|
|
.max_by_key(|ent| {
|
2021-12-07 14:20:45 +00:00
|
|
|
(
|
|
|
|
|
ent.priority,
|
2021-12-08 21:29:08 +00:00
|
|
|
ent.path_prefix
|
2021-12-07 14:20:45 +00:00
|
|
|
.as_ref()
|
|
|
|
|
.map(|x| x.len() as i32)
|
2021-12-08 21:29:08 +00:00
|
|
|
.unwrap_or(0),
|
|
|
|
|
-ent.calls.load(Ordering::SeqCst),
|
2021-12-07 14:20:45 +00:00
|
|
|
)
|
|
|
|
|
});
|
|
|
|
|
|
2021-12-08 21:58:19 +00:00
|
|
|
if let Some(proxy_to) = best_match {
|
|
|
|
|
proxy_to.calls.fetch_add(1, Ordering::SeqCst);
|
2021-12-08 21:24:58 +00:00
|
|
|
|
|
|
|
|
debug!("{}{} -> {}", host, path, proxy_to);
|
2021-12-08 12:28:07 +00:00
|
|
|
trace!("Request: {:?}", req);
|
2021-12-07 14:20:45 +00:00
|
|
|
|
2021-12-08 21:58:19 +00:00
|
|
|
let mut response = if proxy_to.https_target {
|
|
|
|
|
let to_addr = format!("https://{}", proxy_to.target_addr);
|
|
|
|
|
reverse_proxy::call_https(remote_addr.ip(), &to_addr, req).await?
|
|
|
|
|
} else {
|
|
|
|
|
let to_addr = format!("http://{}", proxy_to.target_addr);
|
|
|
|
|
reverse_proxy::call(remote_addr.ip(), &to_addr, req).await?
|
|
|
|
|
};
|
2021-12-07 17:19:51 +00:00
|
|
|
|
|
|
|
|
for (header, value) in proxy_to.add_headers.iter() {
|
2021-12-07 17:50:58 +00:00
|
|
|
response.headers_mut().insert(
|
|
|
|
|
HeaderName::from_bytes(header.as_bytes())?,
|
|
|
|
|
HeaderValue::from_str(value)?,
|
|
|
|
|
);
|
2021-12-07 17:19:51 +00:00
|
|
|
}
|
2021-12-08 12:28:07 +00:00
|
|
|
trace!("Response: {:?}", response);
|
2021-12-08 21:24:58 +00:00
|
|
|
info!("{} {} {}", method, response.status().as_u16(), uri);
|
2021-12-07 17:19:51 +00:00
|
|
|
|
2021-12-09 14:43:19 +00:00
|
|
|
if https_config.enable_compression {
|
|
|
|
|
try_compress(response, accept_encoding, &https_config)
|
|
|
|
|
} else {
|
|
|
|
|
Ok(response)
|
|
|
|
|
}
|
2021-12-07 14:20:45 +00:00
|
|
|
} else {
|
2021-12-08 21:58:19 +00:00
|
|
|
debug!("{}{} -> NOT FOUND", host, path);
|
|
|
|
|
info!("{} 404 {}", method, uri);
|
2021-12-07 14:20:45 +00:00
|
|
|
|
|
|
|
|
Ok(Response::builder()
|
|
|
|
|
.status(StatusCode::NOT_FOUND)
|
|
|
|
|
.body(Body::from("No matching proxy entry"))?)
|
|
|
|
|
}
|
|
|
|
|
}
|
2021-12-09 14:43:19 +00:00
|
|
|
|
|
|
|
|
fn try_compress(
|
|
|
|
|
response: Response<Body>,
|
2021-12-09 15:03:19 +00:00
|
|
|
accept_encoding: Vec<(Option<Encoding>, f32)>,
|
2021-12-09 14:43:19 +00:00
|
|
|
https_config: &HttpsConfig,
|
|
|
|
|
) -> Result<Response<Body>> {
|
2021-12-09 15:03:19 +00:00
|
|
|
let max_q: f32 = accept_encoding
|
|
|
|
|
.iter()
|
|
|
|
|
.max_by_key(|(_, q)| (q * 10000f32) as i64)
|
|
|
|
|
.unwrap_or(&(None, 1.))
|
|
|
|
|
.1;
|
|
|
|
|
let preference = [
|
|
|
|
|
Encoding::Zstd,
|
|
|
|
|
Encoding::Brotli,
|
|
|
|
|
Encoding::Deflate,
|
|
|
|
|
Encoding::Gzip,
|
|
|
|
|
];
|
|
|
|
|
let encoding_opt = accept_encoding
|
|
|
|
|
.iter()
|
|
|
|
|
.filter(|(_, q)| *q == max_q)
|
|
|
|
|
.filter_map(|(enc, _)| *enc)
|
|
|
|
|
.filter(|enc| preference.contains(enc))
|
|
|
|
|
.min_by_key(|enc| preference.iter().position(|x| x == enc).unwrap());
|
|
|
|
|
|
|
|
|
|
// If preferred encoding is none, return as is
|
|
|
|
|
let encoding = match encoding_opt {
|
2021-12-09 14:43:19 +00:00
|
|
|
None | Some(Encoding::Identity) => return Ok(response),
|
|
|
|
|
Some(enc) => enc,
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
// If already compressed, return as is
|
|
|
|
|
if response.headers().get(header::CONTENT_ENCODING).is_some() {
|
|
|
|
|
return Ok(response);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// If content type not in mime types for which to compress, return as is
|
|
|
|
|
match response.headers().get(header::CONTENT_TYPE) {
|
|
|
|
|
Some(ct) => {
|
|
|
|
|
let ct_str = ct.to_str()?;
|
2021-12-09 15:03:19 +00:00
|
|
|
let mime_type = match ct_str.split_once(';') {
|
|
|
|
|
Some((mime_type, _params)) => mime_type,
|
|
|
|
|
None => ct_str,
|
|
|
|
|
};
|
|
|
|
|
if !https_config.compress_mime_types.iter().any(|x| x == mime_type) {
|
2021-12-09 14:43:19 +00:00
|
|
|
return Ok(response);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
None => return Ok(response),
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
debug!("Compressing response body as {:?}", encoding);
|
|
|
|
|
|
|
|
|
|
let (mut head, body) = response.into_parts();
|
|
|
|
|
let body_rd =
|
|
|
|
|
StreamReader::new(body.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)));
|
|
|
|
|
let compressed_body = match encoding {
|
|
|
|
|
Encoding::Gzip => {
|
|
|
|
|
head.headers
|
|
|
|
|
.insert(header::CONTENT_ENCODING, "gzip".parse()?);
|
|
|
|
|
Body::wrap_stream(ReaderStream::new(GzipEncoder::new(body_rd)))
|
|
|
|
|
}
|
|
|
|
|
Encoding::Brotli => {
|
|
|
|
|
head.headers.insert(header::CONTENT_ENCODING, "br".parse()?);
|
|
|
|
|
Body::wrap_stream(ReaderStream::new(BrotliEncoder::new(body_rd)))
|
|
|
|
|
}
|
|
|
|
|
Encoding::Deflate => {
|
|
|
|
|
head.headers
|
|
|
|
|
.insert(header::CONTENT_ENCODING, "deflate".parse()?);
|
|
|
|
|
Body::wrap_stream(ReaderStream::new(DeflateEncoder::new(body_rd)))
|
|
|
|
|
}
|
|
|
|
|
Encoding::Zstd => {
|
|
|
|
|
head.headers
|
|
|
|
|
.insert(header::CONTENT_ENCODING, "zstd".parse()?);
|
|
|
|
|
Body::wrap_stream(ReaderStream::new(ZstdEncoder::new(body_rd)))
|
|
|
|
|
}
|
|
|
|
|
_ => unreachable!(),
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
Ok(Response::from_parts(head, compressed_body))
|
|
|
|
|
}
|
