Refactor proxy

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 resolver = "2"
 members = [
+    "dynamic-proxy",
     "plane/plane-tests",
     "plane/plane-dynamic",
     "plane",
@@ -12,5 +13,6 @@ members = [
 # https://github.com/rust-lang/cargo/pull/9252/files
 default-members = [
     "plane",
-    "plane/plane-tests"
+    "plane/plane-tests",
+    "dynamic-proxy",
 ]

LICENSE

@@ -19,3 +19,27 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
+
+---
+
+Contains code from hyperium/hyper-util, licensed under the MIT license:
+
+Copyright (c) 2023 Sean McArthur
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

dynamic-proxy/.gitignore

@@ -0,0 +1 @@
+/target

dynamic-proxy/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "dynamic-proxy"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = "1.0.89"
+bytes = "1.7.2"
+http = "1.1.0"
+http-body = "1.0.1"
+http-body-util = "0.1.2"
+hyper = "1.4.1"
+hyper-util = { version = "0.1.8", features = ["http1", "http2", "server", "server-graceful", "server-auto", "client", "client-legacy"] }
+pin-project-lite = "0.2.14"
+rustls = { version = "0.23.13", features = ["ring"] }
+thiserror = "1.0.63"
+serde = { version = "1.0.210", features = ["derive"] }
+tokio = { version = "1.40.0", features = ["macros", "rt-multi-thread"] }
+tokio-rustls = "0.26.0"
+tracing = "0.1.40"
+
+[dev-dependencies]
+axum = { version = "0.7.6", features = ["http2", "ws"] }
+futures-util = "0.3.30"
+http = "1.1.0"
+rcgen = "0.13.1"
+reqwest = { version = "0.12.7", features = ["http2", "stream"] }
+serde_json = "1.0.128"
+tokio-tungstenite = "0.24.0"

dynamic-proxy/src/body.rs

@@ -0,0 +1,24 @@
+//! Provides a concrete, boxed body and error type.
+
+use bytes::Bytes;
+use http_body::Body;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Empty};
+
+pub type BoxedError = Box<dyn std::error::Error + Send + Sync>;
+
+pub type SimpleBody = BoxBody<Bytes, BoxedError>;
+
+pub fn to_simple_body<B>(body: B) -> SimpleBody
+where
+    B: Body<Data = Bytes> + Send + Sync + 'static,
+    B::Error: Into<BoxedError>,
+{
+    body.map_err(|e| e.into() as BoxedError).boxed()
+}
+
+pub fn simple_empty_body() -> SimpleBody {
+    Empty::<Bytes>::new()
+        .map_err(|_| unreachable!("Infallable"))
+        .boxed()
+}

dynamic-proxy/src/graceful_shutdown.rs

@@ -0,0 +1,101 @@
+//! Near-identical copy of hyper_util::server::graceful::GracefulShutdown
+//! that derives `Clone` and adds a `subscribe` method.
+//! https://github.com/hyperium/hyper-util/blob/master/src/server/graceful.rs
+
+use hyper_util::server::graceful::GracefulConnection;
+use pin_project_lite::pin_project;
+use std::{
+    fmt::{self, Debug},
+    future::Future,
+    pin::Pin,
+    task::{self, Poll},
+};
+use tokio::sync::watch;
+
+#[derive(Clone)] // Added in Plane
+pub struct GracefulShutdown {
+    tx: watch::Sender<()>,
+}
+
+impl GracefulShutdown {
+    /// Create a new graceful shutdown helper.
+    pub fn new() -> Self {
+        let (tx, _) = watch::channel(());
+        Self { tx }
+    }
+
+    /// Wrap a future for graceful shutdown watching.
+    pub fn watch<C: GracefulConnection>(&self, conn: C) -> impl Future<Output = C::Output> {
+        let mut rx = self.tx.subscribe();
+        GracefulConnectionFuture::new(conn, async move {
+            let _ = rx.changed().await;
+            // hold onto the rx until the watched future is completed
+            rx
+        })
+    }
+
+    // Added in Plane
+    pub fn subscribe(&self) -> watch::Receiver<()> {
+        self.tx.subscribe()
+    }
+
+    /// Signal shutdown for all watched connections.
+    ///
+    /// This returns a `Future` which will complete once all watched
+    /// connections have shutdown.
+    pub async fn shutdown(self) {
+        let Self { tx } = self;
+
+        // signal all the watched futures about the change
+        let _ = tx.send(());
+        // and then wait for all of them to complete
+        tx.closed().await;
+    }
+}
+
+pin_project! {
+    struct GracefulConnectionFuture<C, F: Future> {
+        #[pin]
+        conn: C,
+        #[pin]
+        cancel: F,
+        #[pin]
+        // If cancelled, this is held until the inner conn is done.
+        cancelled_guard: Option<F::Output>,
+    }
+}
+
+impl<C, F: Future> GracefulConnectionFuture<C, F> {
+    fn new(conn: C, cancel: F) -> Self {
+        Self {
+            conn,
+            cancel,
+            cancelled_guard: None,
+        }
+    }
+}
+
+impl<C, F: Future> Debug for GracefulConnectionFuture<C, F> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GracefulConnectionFuture").finish()
+    }
+}
+
+impl<C, F> Future for GracefulConnectionFuture<C, F>
+where
+    C: GracefulConnection,
+    F: Future,
+{
+    type Output = C::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+        if this.cancelled_guard.is_none() {
+            if let Poll::Ready(guard) = this.cancel.poll(cx) {
+                this.cancelled_guard.set(Some(guard));
+                this.conn.as_mut().graceful_shutdown();
+            }
+        }
+        this.conn.poll(cx)
+    }
+}

dynamic-proxy/src/https_redirect.rs

@@ -0,0 +1,70 @@
+use crate::body::{simple_empty_body, BoxedError, SimpleBody};
+use http::{
+    header,
+    uri::{Authority, Scheme},
+    Request, Response, StatusCode, Uri,
+};
+use hyper::{body::Incoming, service::Service};
+use std::{future::ready, pin::Pin, str::FromStr};
+
+/// A hyper service that redirects HTTP requests to HTTPS.
+#[derive(Debug, Clone)]
+pub struct HttpsRedirectService;
+
+impl HttpsRedirectService {
+    fn call_inner(request: Request<Incoming>) -> Result<Response<SimpleBody>, StatusCode> {
+        // Get the host header.
+        let hostname = request
+            .headers()
+            .get(header::HOST)
+            .ok_or(StatusCode::BAD_REQUEST)?;
+        // Parse the host header into an authority.
+        let authority =
+            Authority::from_str(hostname.to_str().map_err(|_| StatusCode::BAD_REQUEST)?)
+                .map_err(|_| StatusCode::BAD_REQUEST)?;
+        // Strip the port.
+        let authority =
+            Authority::from_str(authority.host()).expect("Valid host is always valid authority.");
+
+        let request_uri = request.uri().clone();
+
+        // Set the scheme to HTTPS
+        let mut parts = request_uri.into_parts();
+        parts.scheme = Some(Scheme::HTTPS);
+
+        parts.authority = Some(authority);
+
+        // Build the new URI
+        let new_uri = Uri::from_parts(parts).expect("URI is always valid");
+
+        let response = Response::builder()
+            .status(StatusCode::MOVED_PERMANENTLY)
+            .header(header::LOCATION, new_uri.to_string())
+            .body(simple_empty_body());
+
+        response.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
+    }
+}
+
+impl Service<Request<Incoming>> for HttpsRedirectService {
+    type Response = Response<SimpleBody>;
+    type Error = BoxedError;
+    type Future = Pin<Box<std::future::Ready<Result<Response<SimpleBody>, BoxedError>>>>;
+
+    fn call(&self, request: Request<Incoming>) -> Self::Future {
+        let result = Self::call_inner(request);
+
+        let result = match result {
+            Ok(response) => response,
+            Err(status) => {
+                tracing::error!("Error redirecting to HTTPS: {}", status);
+                Response::builder()
+                    .status(status)
+                    .body(simple_empty_body())
+                    .expect("Response is always valid")
+            }
+        };
+
+        Box::pin(ready(Ok(result)))
+    }
+}

dynamic-proxy/src/lib.rs

@@ -0,0 +1,11 @@
+pub mod body;
+mod graceful_shutdown;
+pub mod https_redirect;
+pub mod proxy;
+pub mod request;
+pub mod server;
+mod upgrade;
+
+pub use hyper;
+pub use rustls;
+pub use tokio_rustls;