Refactor proxy (#814)

This refactors the reverse proxy embedded in Plane.

The main goal of this is to update hyper support to the `1.x` branch. As
side-goals, it:
- Refactors the proxy substantially to separate proxy logic from Plane
business logic
- Adds a bunch of tests and some testing infrastructure

Rather than upgrading the entire `plane` project to the latest version
of hyper, this PR introduces a `dynamic-proxy` crate which contains the
HTTP-level proxy implementation using the latest stable version of each
of its dependencies. This allows us to decouple the dependency versions
of the proxy from the rest of Plane.

Once this is merged, the next step will be to update the `axum` and
`reqwests` crates in `plane` itself so that they use the latest `hyper`.
Then, we can either keep `dynamic-proxy` as a separate crate and publish
it (e.g. as `plane-dynamic-proxy`), or merge it back in to the main
`plane` crate.

[Requirements
matrix](https://docs.google.com/document/d/1cxc-8xamW6BzoflsQmKkW8pwKm8USVD7Tq1PZnpoR_4/edit)

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;