feat: implement TLS for proxy and mgmt servers

certs/README.md

@@ -0,0 +1,17 @@
+# TLS Certs
+
+> [!CAUTION]
+> These are intended for development only!
+
+This directory contains TLS certs for localhost. These are used to make development easier.
+
+## Making your own certificates
+
+```
+openssl req -x509 -out localhost.crt -keyout localhost.key \
+  -newkey rsa:2048 -nodes -sha256 \
+  -subj '/CN=localhost' -extensions EXT -config <( \
+   printf "[dn]\nCN=localhost\n[req]\ndistinguished_name = dn\n[EXT]\nsubjectAltName=DNS:localhost\nkeyUsage=digitalSignature\nextendedKeyUsage=serverAuth")
+```
+
+[source](https://letsencrypt.org/docs/certificates-for-localhost/#making-and-trusting-your-own-certificates)

certs/localhost.crt

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDDzCCAfegAwIBAgIUWSq6qNcWO52SspWx9KyPKMaS/UkwDQYJKoZIhvcNAQEL
+BQAwFDESMBAGA1UEAwwJbG9jYWxob3N0MB4XDTI0MDEyODEzNDcyNFoXDTI0MDIy
+NzEzNDcyNFowFDESMBAGA1UEAwwJbG9jYWxob3N0MIIBIjANBgkqhkiG9w0BAQEF
+AAOCAQ8AMIIBCgKCAQEAoyXbnny+Plmo0syJX7r5R0FSlV0NQWmfapKyi3ChZ6ez
+H4e21SXAo0UPOvBFUfp+Gj1t2uTP86XGP8tN59/T1iI8Rvzy7bmfm433Ql2JoUFN
+iOjqQYyIH6cepM160nXjupS5gbtWbO2Q/4uYgqcPQegK3gKXkA9rKQmj6toRAyBy
+e8lxrE+ULQdjxifjlP5IFpukyWdlq+Meuza+Hzzrp/x2EHsidzhNZvwfO8HAB/Ej
+phL6bnmQM0ZkmVhoq5L8F3RnrZlw/I62qp6fq8iF64rZoJtuyvg6xu8UhaCHmI4J
+bhA6i56UFBDrZ9Om37zxyMlzKxePoeZh0ieVw5ANMwIDAQABo1kwVzAUBgNVHREE
+DTALgglsb2NhbGhvc3QwCwYDVR0PBAQDAgeAMBMGA1UdJQQMMAoGCCsGAQUFBwMB
+MB0GA1UdDgQWBBS/imYFn4ntU1xD5Z5lUfYSVigy7TANBgkqhkiG9w0BAQsFAAOC
+AQEAkzZCmv49YZfIHEDEl84pKMLKO81FY9l+0lhB6y9F1fV93Ch8EByEvmBG6G5m
+mdfJi7+4LGVkPNsMs0eD6b4MUmV71pHl+NSKtXuMc2YS8oAL8g6+IFDQAZhl0Mpo
+bElR86QvW3AoiC1QDJUFzfVC0WMMRA7YmDzWYyJX9H/zPWPyNvkvAppeVGBj+f1V
+AaGcg6yBpM9XZB2jYkIXiPO1J+/X0YnMqn6RqF+Zg+nZROPNwYVzn+TUcffwP47D
+az/Itnlh0t3aWQ2rI7NkjPDJQ4FJUfseAWd244un8IE0MO6PLvD1hSWiMG4Dn0BE
+j3xFGLw7g67KwKxl9rVDAoPGBg==
+-----END CERTIFICATE-----

certs/localhost.key

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCjJduefL4+WajS
+zIlfuvlHQVKVXQ1BaZ9qkrKLcKFnp7Mfh7bVJcCjRQ868EVR+n4aPW3a5M/zpcY/
+y03n39PWIjxG/PLtuZ+bjfdCXYmhQU2I6OpBjIgfpx6kzXrSdeO6lLmBu1Zs7ZD/
+i5iCpw9B6AreApeQD2spCaPq2hEDIHJ7yXGsT5QtB2PGJ+OU/kgWm6TJZ2Wr4x67
+Nr4fPOun/HYQeyJ3OE1m/B87wcAH8SOmEvpueZAzRmSZWGirkvwXdGetmXD8jraq
+np+ryIXritmgm27K+DrG7xSFoIeYjgluEDqLnpQUEOtn06bfvPHIyXMrF4+h5mHS
+J5XDkA0zAgMBAAECggEAEHFHUhDGZ6hHoH8mtTQ13V2TAiSvqlEH1QjV38HJMpYv
+MlVOMussIAhcwZbnlZyGSwS35qC66JZjhZhq0Jy5T9KBerIRla3ojRfgvJqKvWrE
+crDusw6DxZTlPLzMzRs+iVZl39JOoonK3EZoZ3qIyh6lwbxHJCi5pxgmogu6PTNF
+nf8zULHWP/8D+vsVWWY7scnZbH5Z/6L0fodeNYnsfs4lh9dStbqs0wKVrcbQnPW7
+JYpqdJ0Rr5KpWlrPvqZlzUIlkryihZBhIl5eRIa49fOOcliSxH8QWtBczBYTf/Y8
+EiPxcEBURxx9q2SruImhFmkI6T42u2rUtFGyaUFswQKBgQDerRB05P5NBGM+f9PN
+cV2mQKAqyHyzq6srLQo2a0X//6xN+NsvtBasm2hOYWSg9+wV2/yeF0TA/2UBFkOh
+Iza1l0Fb0MjQGGJLH0YDfSw004v54tahhQQR3l3BI5qBUHjNQDfwRIG1/JaOfJ5l
+Y8xUi1I0OT0lM0soBOl3VAIw4QKBgQC7kDZSAzt1hTmZbyOCX3yfMw8oXWc/ATpj
+WtzuJTY2PUal00yYm58bUrZmhj0XeYHQOvgXJ9KzMbJMzIWt20xUAIflL6JDZebw
+Eceq1j3X6JLqhRpgUgOhVf5yLbRWbsY0h3Q7YxEQf90W4vpgj3XrVkQ6KqsXsml0
+g9tzPcF8kwKBgQDMJRfoQyRNEY+29dQFDkDQMYFll8aTpffYLoOlXnWffBPIrDSu
+qEj9V8Cp0ypBVOnRJIyVlzmGQt6jv3ijGziGBLR7646fESvUOUij3DcR+zviDS++
+hsczZozHi8+TbGZDrfNayEOux3J0ERXaWEM040Gq9Sr0lvD5MH+l0ZPsoQKBgB0Q
+JIqiu5TjNuCiiwMJnrrgY4nipzvpCc4ZZ0Bzfan75rWNP0IqYwYN0/ug81hu2IGW
+kZis8AYaPkGOM2yUHYiqqGQH9IGzCYzLhH/hQKXzAMjcJRElxDA8rfetQ1NdSNMc
+5hLJr/w5g92nABr0P9ZegKXutKIwYAzQ3bFGsXOHAoGAdo2A6Ug9wuGUDrJjSSRW
+6B3DomKUB47OQbsOjLi0VY8jTp1YJ1WD4ZV7sjgDjcTeit7g5B5dkoLttwT4/ctj
+Zz/4YAmtsJTUxIpk8b2jT79HeWrSrT+zmF/zoIGkx9dCks01dh+DfCQeNHx9PDGz
+dAGBSME1eIbMfXkKEYbDpI4=
+-----END PRIVATE KEY-----

crates/proxy/Cargo.toml

@@ -6,7 +6,8 @@ default-run = "soldr"
 
 [dependencies]
 anyhow = "1.0"
-axum = "0.6.18"
+axum = "0.7"
+axum-server = { version = "0.6", features = ["tls-rustls"] }
 clap = { version = "4.3.8", features = ["derive"] }
 hyper = { version = "0.14", features = ["full"] }
 lettre = { version = "0.10.4", default-features = false, features = ["smtp-transport", "tokio1", "tokio1-rustls-tls", "builder"] }
@@ -21,7 +22,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tokio = { version = "1.0", features = ["full"] }
 toml = "0.7.5"
 tower = { version = "0.4", features = ["util"] }
-tower-http = { version = "0.4.0", features = ["trace", "cors"] }
+tower-http = { version = "0.5", features = ["trace", "cors"] }
 
 [dev-dependencies]
 criterion = {version = "0.4", features = ["async_tokio"]}

crates/proxy/examples/origin.rs

@@ -1,6 +1,7 @@
 use anyhow::Result;
 use axum::http::StatusCode;
 use axum::{routing::any, Router};
+use tokio::net::TcpListener;
 use tokio::time::{sleep, Duration};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -20,10 +21,9 @@ async fn main() -> Result<()> {
         .route("/timeout", any(timeout_handler));
 
     let addr = "0.0.0.0:8080";
+    let listener = TcpListener::bind(addr).await?;
     tracing::info!("origin listening on {}", addr);
-    axum::Server::bind(&addr.parse()?)
-        .serve(origin.into_make_service())
-        .await?;
+    axum::serve(listener, origin).await?;
 
     Ok(())
 }

crates/proxy/src/lib.rs

@@ -15,11 +15,9 @@ use std::result::Result as StdResult;
 use anyhow::Result;
 use axum::body::Body;
 use axum::extract::{Extension, State};
-use axum::http::Request;
-use axum::http::StatusCode;
+use axum::http::{HeaderMap, Request, StatusCode};
 use axum::response::IntoResponse;
 use axum::{routing::any, Router};
-use hyper::HeaderMap;
 use queue::RetryQueue;
 use serde::Deserialize;
 use sqlx::sqlite::SqlitePool;
@@ -32,12 +30,21 @@ use crate::proxy::{proxy, Client};
 use crate::request::HttpRequest;
 use crate::request::State as RequestState;
 
+#[derive(Debug, Default, Deserialize)]
+#[serde(default)]
+pub struct Tls {
+    pub enable: bool,
+    pub cert_path: Option<String>,
+    pub key_path: Option<String>,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(default)]
 pub struct Config {
     pub database_url: String,
     pub management_listener: String,
     pub ingest_listener: String,
+    pub tls: Tls,
 }
 
 impl Default for Config {
@@ -48,6 +55,11 @@ impl Default for Config {
             database_url: "sqlite::memory:".to_string(),
             management_listener: "0.0.0.0:3443".to_string(),
             ingest_listener: "0.0.0.0:3000".to_string(),
+            tls: Tls {
+                enable: false,
+                cert_path: None,
+                key_path: None,
+            },
         }
     }
 }
@@ -85,7 +97,7 @@ async fn handler(
     let uri = req.uri().to_string();
     let headers = transform_headers(req.headers());
     let body = req.into_body();
-    let body = hyper::body::to_bytes(body).await?;
+    let body = axum::body::to_bytes(body, 1_000_000).await?;
     let r = HttpRequest {
         method,
         uri,

crates/proxy/src/main.rs

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use axum_server::tls_rustls::RustlsConfig;
 use clap::Parser;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -32,7 +33,7 @@ async fn main() -> Result<()> {
 
     tokio::spawn(async move {
         tracing::info!("management API listening on {}", mgmt_listener);
-        if let Err(err) = axum::Server::bind(&mgmt_listener)
+        if let Err(err) = axum_server::bind(mgmt_listener)
             .serve(mgmt.into_make_service())
             .await
         {
@@ -45,10 +46,25 @@ async fn main() -> Result<()> {
         retry_queue.start().await;
     });
 
+    let tls_config = if config.tls.enable {
+        let cert_path = config.tls.cert_path.unwrap();
+        let key_path = config.tls.key_path.unwrap();
+        Some(RustlsConfig::from_pem_file(cert_path, key_path).await?)
+    } else {
+        None
+    };
+
     tracing::info!("ingest listening on {}", ingest_listener);
-    axum::Server::bind(&ingest_listener)
-        .serve(ingest.into_make_service())
-        .await?;
+    if let Some(tls_config) = tls_config {
+        tracing::info!("tls configured for {}", ingest_listener);
+        axum_server::bind_rustls(ingest_listener, tls_config)
+            .serve(ingest.into_make_service())
+            .await?;
+    } else {
+        axum_server::bind(ingest_listener)
+            .serve(ingest.into_make_service())
+            .await?;
+    }
 
     Ok(())
 }

crates/proxy/src/origin.rs

@@ -1,4 +1,4 @@
-use axum::http::Uri;
+use hyper::Uri;
 
 pub struct Origin {
     pub uri: Uri,

crates/proxy/src/proxy.rs

@@ -1,9 +1,6 @@
 use anyhow::{anyhow, Result};
-use axum::http::Request;
-use axum::http::Uri;
 use hyper::client::HttpConnector;
-use hyper::Body;
-use hyper::Response;
+use hyper::{Body, Request, Response, Uri};
 use sqlx::SqlitePool;
 use tokio::time::{timeout, Duration};
 

crates/proxy/tests/integration/common.rs

@@ -27,5 +27,6 @@ pub fn config() -> Config {
         database_url: "sqlite::memory:".to_string(),
         management_listener: "0.0.0.0:3443".to_string(),
         ingest_listener: "0.0.0.0:3000".to_string(),
+        tls: Default::default(),
     }
 }

crates/proxy/tests/integration/ingest.rs

@@ -1,6 +1,5 @@
 use crate::common;
 
-use std::net::{SocketAddr, TcpListener};
 use std::sync::Arc;
 
 use axum::body::Body;
@@ -9,6 +8,7 @@ use axum::http::Request;
 use axum::http::StatusCode;
 use axum::{routing::post, Router};
 use soldr::db::RequestState;
+use tokio::net::TcpListener;
 use tokio::sync::Mutex;
 use tokio::time::{sleep, Duration};
 use tower::util::ServiceExt;
@@ -42,18 +42,14 @@ async fn timeout_handler() -> impl axum::response::IntoResponse {
 #[tokio::test]
 async fn ingest_save_and_proxy() {
     // set up origin server
-    let listener = TcpListener::bind("0.0.0.0:0".parse::<SocketAddr>().unwrap()).unwrap();
+    let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
     let port = listener.local_addr().unwrap().port();
     let sentinel: Sentinel = Arc::new(Mutex::new(None));
     let s2 = sentinel.clone();
     let client_app = Router::new().route("/", post(success_handler).with_state(s2));
 
     tokio::spawn(async move {
-        axum::Server::from_tcp(listener)
-            .unwrap()
-            .serve(client_app.into_make_service())
-            .await
-            .unwrap();
+        axum::serve(listener, client_app).await.unwrap();
     });
 
     let (ingest, mgmt, _) = app(&common::config()).await.unwrap();
@@ -74,7 +70,7 @@ async fn ingest_save_and_proxy() {
                 .method("POST")
                 .uri("/origins")
                 .header("Content-Type", "application/json")
-                .body(body.into())
+                .body(body)
                 .unwrap(),
         )
         .await
@@ -117,7 +113,9 @@ async fn ingest_save_and_proxy() {
 
     assert_eq!(response.status(), StatusCode::OK);
 
-    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
+    let body = axum::body::to_bytes(response.into_body(), 1_000_000)
+        .await
+        .unwrap();
 
     let reqs: Vec<db::Request> = serde_json::from_slice(&body).unwrap();
     assert_eq!(reqs[0].state, RequestState::Completed);
@@ -137,7 +135,9 @@ async fn ingest_save_and_proxy() {
 
     assert_eq!(response.status(), StatusCode::OK);
 
-    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
+    let body = axum::body::to_bytes(response.into_body(), 1_000_000)
+        .await
+        .unwrap();
 
     let attempts: Vec<db::Attempt> = serde_json::from_slice(&body).unwrap();
     assert_eq!(attempts[0].id, 1);
@@ -154,16 +154,12 @@ async fn ingest_proxy_failure() {
     common::enable_tracing();
 
     // set up origin server
-    let listener = TcpListener::bind("0.0.0.0:0".parse::<SocketAddr>().unwrap()).unwrap();
+    let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
     let port = listener.local_addr().unwrap().port();
     let client_app = Router::new().route("/failure", post(failure_handler));
 
     tokio::spawn(async move {
-        axum::Server::from_tcp(listener)
-            .unwrap()
-            .serve(client_app.into_make_service())
-            .await
-            .unwrap();
+        axum::serve(listener, client_app).await.unwrap();
     });
 
     let (ingest, mgmt, _) = app(&common::config()).await.unwrap();
@@ -190,7 +186,7 @@ async fn ingest_proxy_failure() {
                 .method("POST")
                 .uri("/origins")
                 .header("Content-Type", "application/json")
-                .body(body.into())
+                .body(body)
                 .unwrap(),
         )
         .await
@@ -231,7 +227,9 @@ async fn ingest_proxy_failure() {
 
     assert_eq!(response.status(), StatusCode::OK);
 
-    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
+    let body = axum::body::to_bytes(response.into_body(), 1_000_000)
+        .await
+        .unwrap();
 
     let reqs: Vec<db::Request> = serde_json::from_slice(&body).unwrap();
     assert_eq!(reqs[0].state, RequestState::Failed);
@@ -251,7 +249,9 @@ async fn ingest_proxy_failure() {
 
     assert_eq!(response.status(), StatusCode::OK);
 
-    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
+    let body = axum::body::to_bytes(response.into_body(), 1_000_000)
+        .await
+        .unwrap();
 
     let attempts: Vec<db::Attempt> = serde_json::from_slice(&body).unwrap();
     assert_eq!(attempts[0].id, 1);
@@ -265,16 +265,12 @@ async fn ingest_proxy_timeout() {
     common::enable_tracing();
 
     // set up origin server
-    let listener = TcpListener::bind("0.0.0.0:0".parse::<SocketAddr>().unwrap()).unwrap();
+    let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
     let port = listener.local_addr().unwrap().port();
     let client_app = Router::new().route("/timeout", post(timeout_handler));
 
     tokio::spawn(async move {
-        axum::Server::from_tcp(listener)
-            .unwrap()
-            .serve(client_app.into_make_service())
-            .await
-            .unwrap();
+        axum::serve(listener, client_app).await.unwrap();
     });
 
     let (ingest, mgmt, _) = app(&common::config()).await.unwrap();
@@ -295,7 +291,7 @@ async fn ingest_proxy_timeout() {
                 .method("POST")
                 .uri("/origins")
                 .header("Content-Type", "application/json")
-                .body(body.into())
+                .body(body)
                 .unwrap(),
         )
         .await
@@ -336,7 +332,9 @@ async fn ingest_proxy_timeout() {
 
     assert_eq!(response.status(), StatusCode::OK);
 
-    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
+    let body = axum::body::to_bytes(response.into_body(), 1_000_000)
+        .await
+        .unwrap();
 
     let reqs: Vec<db::Request> = serde_json::from_slice(&body).unwrap();
     assert_eq!(reqs[0].state, RequestState::Timeout);
@@ -356,7 +354,9 @@ async fn ingest_proxy_timeout() {
 
     assert_eq!(response.status(), StatusCode::OK);
 
-    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
+    let body = axum::body::to_bytes(response.into_body(), 1_000_000)
+        .await
+        .unwrap();
 
     let attempts: Vec<db::Attempt> = serde_json::from_slice(&body).unwrap();
     assert_eq!(attempts[0].id, 1);