Reverse proxy & feedback

[antoniosarosi]

I've been working on a little reverse proxy project over the last couple weeks, nothing serious, just for fun. Here are some issues I found specific to hyper and this kind of projects:

HTTP/1.1 Upgraded connections

There is an example of how connection upgrades work with hyper here, but it doesn't apply to reverse proxies, since they don't handle the upgraded connection themselves. Instead, they have to set up a tunnel and forward data from the client to the upstream server and viceversa. There is also an example of how to set up a tunnel here, this is used with the CONNECT header by a forward proxy. The problem with a reverse proxy is that it cannot immediately upgrade the connection, it has to send the request to the upstream server and wait for the HTTP 101 response, then upgrade the connection and send that response back to the client.

The examples use hyper::upgrade::on to upgrade on either a request or response. The issue is that upgrading on a request as a server requires giving up ownership on that request, because you need to spawn a Tokio task since the hyper::upgrade::on future will not resolve until you send an HTTP 101 response back to the client. From this example:

// Note: This can't possibly be fulfilled until the 101 response
// is returned below, so it's better to spawn this future instead
// waiting for it to complete to then return a response.
tokio::task::spawn(async move {
    match hyper::upgrade::on(&mut req).await {
        Ok(upgraded) => {
            if let Err(e) = server_upgraded_io(upgraded).await {
                eprintln!("server foobar io error: {}", e)
            };
        }
        Err(e) => eprintln!("upgrade error: {}", e),
    }
});

As a reverse proxy, you have to forward the incoming request to the upstream server and also upgrade on it, which you can't do easily because for both actions you need ownership on the request. So this is more or less what I ended up doing:

let mut maybe_client_upgrade = None;

if request.headers().contains_key(header::UPGRADE) {
    let upgrade = request.extensions_mut().remove::<OnUpgrade>().unwrap();
    maybe_client_upgrade = Some(upgrade);
}

let mut response = sender.send_request(request).await?;

if response.status() == http::StatusCode::SWITCHING_PROTOCOLS {
    if let Some(client_upgrade) = maybe_client_upgrade {
        let server_upgrade = response.extensions_mut().remove::<OnUpgrade>().unwrap();
        tokio::task::spawn(proxy_tunnel(client_upgrade, server_upgrade));
    } else {
        // Upstream server sent us an HTTP 101 response without the client
        // asking for an upgrade, so we can't proxy data from the client.
        return Ok(LocalResponse::bad_gateway());
    }
}

I found out while reading through the source code that hyper::upgrade::on basically calls this function:

impl<B> CanUpgrade for http::Request<B> {
    fn on_upgrade(mut self) -> OnUpgrade {
        self.extensions_mut()
            .remove::<OnUpgrade>()
            .unwrap_or_else(OnUpgrade::none)
    }
}

So I figured out I can do that manually and then await the OnUpgrade futures like this:

async fn proxy_tunnel(client: OnUpgrade, server: OnUpgrade) {
    let (mut upgraded_client, mut upgraded_server) = tokio::try_join!(client, server).unwrap();

    match tokio::io::copy_bidirectional(&mut upgraded_client, &mut upgraded_server).await {
        Ok((client_bytes, server_bytes)) => {
            println!("Client wrote {client_bytes} bytes, server wrote {server_bytes} bytes")
        }
        Err(err) => eprintln!("Tunnel error: {err}"),
    }
}

It works, but I'm not sure if it's okay to remove that extension from the request and response.

No access to the underlying socket

I implemented the Forwarded HTTP header, for which I need to know my listening address and the client address. In previous versions (I'm using 1.0.0-rc2) there was a method remote_addr() on the request or an extension type, but it's not available anymore. I found this response to an issue from 2018, so that's what I did, I just made my own struct that implements Service:

pub(crate) struct Proxy {
    /// Reference to global config.
    config: &'static Config,

    // Socket address of the connected client.
    client_addr: SocketAddr,

    // Listening socket address.
    server_addr: SocketAddr,
}

This is not a really a problem but it adds boilerplate, because if I could get the socket addresses from the request I wouldn't have to implement Service, I could use service_fn and a function.

Body types

Other than that everything was great, I just missed some documentation on body types as some examples (http_proxy, web_api, echo) tend to use BoxBody but they don't explain why. I'm assuming it's because BoxBody can represent both an incoming body and a complete body, so it's useful for echoing requests or sending a custom 404 response under the same return type. I'll update this discussion if I find more issues while implementing new features.