[RFC pve-http-server 01/13] anyevent : proxy a path prefix to a local http proxy

[Alexandre Derumier <alexandre.derumier@groupe-cyllene.com>]

A handler says which prefixes go to a local backend and who may reach them,
so a service can use this server's TLS and authentication without its own port.
Requests go verbatim, not re-encoded like proxy_request; an upgrade becomes a
pipe after 101, and the relay stops reading while the far side is behind.

Signed-off-by: Alexandre Derumier <alexandre.derumier@groupe-cyllene.com>
---
 src/PVE/APIServer/AnyEvent.pm | 263 ++++++++++++++++++++++++++++++++++
 1 file changed, 263 insertions(+)

diff --git a/src/PVE/APIServer/AnyEvent.pm b/src/PVE/APIServer/AnyEvent.pm
index 915d678..dc95c12 100644
--- a/src/PVE/APIServer/AnyEvent.pm
+++ b/src/PVE/APIServer/AnyEvent.pm
@@ -731,6 +731,250 @@ sub websocket_proxy {
     }
 }
 
+# Queued for one side before the other stops being read. Smaller than
+# response_stream's 4MB: consoles are not downloads, and a backlog is only
+# latency the far end could have coalesced away.
+my $relay_buf_size = 1024 * 1024;
+
+# What a handle still owes its socket; TLS keeps a second buffer.
+sub relay_queued {
+    my ($hdl) = @_;
+    return length($hdl->{wbuf}) + length($hdl->{_tls_wbuf} // '');
+}
+
+# One direction of an upgraded connection: copy to the other side, and stop
+# reading while that side is behind, so back pressure reaches the far end
+# instead of queueing here. Same shape as response_stream, and named rather
+# than a closure over itself, which would be a cycle. The handles come from
+# callbacks because either may be gone by the time this runs.
+sub relay_reader {
+    my ($from, $to) = @_;
+
+    return sub {
+        my ($hdl) = @_;
+
+        my $writer = $to->();
+        return if !$writer;
+
+        my $data = $hdl->{rbuf};
+        $hdl->{rbuf} = '';
+        $writer->push_write($data) if length($data);
+
+        return if relay_queued($writer) < $relay_buf_size;
+
+        my $prev_on_drain = $writer->{on_drain};
+        $writer->on_drain(sub {
+            my ($wrhdl) = @_;
+            # Restored first: setting on_drain runs it on an empty buffer.
+            $wrhdl->on_drain($prev_on_drain);
+            if (my $reader = $from->()) {
+                $reader->on_read(relay_reader($from, $to));
+            }
+        });
+
+        $hdl->on_read();
+    };
+}
+
+# Hand an upgrade to the backend: the request goes out as it arrived and the
+# answer comes back untouched, so the two ends compute the accept key. After
+# 101 this is a pipe, which knows nothing of websockets.
+sub local_http_proxy_upgrade {
+    my ($self, $reqstate, $method, $target) = @_;
+
+    my $r = $reqstate->{request};
+
+    my ($remhost, $remport);
+    if ($target->{port}) {
+        $remhost = 'localhost';
+        $remport = $target->{port};
+    } else {
+        $remhost = 'unix/';
+        $remport = $target->{socket};
+    }
+    my $path = $target->{path} // '/';
+
+    # Only Host is rewritten: this is the hop being upgraded, so Connection
+    # and Upgrade stay.
+    my $headers = '';
+    $r->headers->scan(sub {
+        my ($key, $value) = @_;
+        return if lc($key) eq 'host';
+        $headers .= "$key: $value\015\012";
+    });
+    my $request = "$method $path HTTP/1.1\015\012Host: localhost\015\012$headers\015\012";
+
+    tcp_connect $remhost, $remport, sub {
+        my ($fh) = @_
+            or do {
+                $self->error($reqstate, HTTP_BAD_GATEWAY, "connect to backend failed: $!");
+                return;
+            };
+
+        $reqstate->{proxyhdl} = AnyEvent::Handle->new(
+            fh => $fh,
+            rbuf_max => 64 * 1024,
+            wbuf_max => 4 * $relay_buf_size,
+            timeout => 30,
+            on_eof => sub {
+                eval {
+                    $self->log_aborted_request($reqstate);
+                    $self->client_do_disconnect($reqstate);
+                };
+                warn $@ if $@;
+            },
+            on_error => sub {
+                my ($hdl, $fatal, $message) = @_;
+                eval {
+                    $self->log_aborted_request($reqstate, $message);
+                    $self->client_do_disconnect($reqstate);
+                };
+                warn $@ if $@;
+            },
+        );
+
+        $reqstate->{proxyhdl}->push_write($request);
+
+        $reqstate->{proxyhdl}->push_read(
+            line => "\015\012\015\012",
+            sub {
+                my ($hdl, $response) = @_;
+
+                # Only 101 means the backend stopped speaking HTTP.
+                if ($response !~ m|^HTTP/1\.1 101|) {
+                    my ($status) = $response =~ m|^(\S+ \d+[^\015]*)|;
+                    $self->log_aborted_request($reqstate,
+                        "backend refused upgrade: " . ($status // 'unparseable response'));
+                    $self->client_do_disconnect($reqstate);
+                    return;
+                }
+
+                # Verbatim: it carries the accept key for the client's key.
+                $reqstate->{hdl}->push_write($response . "\015\012\015\012");
+
+                $reqstate->{proxyhdl}->timeout(0);
+                $reqstate->{hdl}->timeout(0);
+
+                my $client = sub { $reqstate->{hdl} };
+                my $backend = sub { $reqstate->{proxyhdl} };
+
+                $reqstate->{proxyhdl}->on_read(relay_reader($backend, $client));
+                $reqstate->{hdl}->on_read(relay_reader($client, $backend));
+
+                $reqstate->{log}->{code} = 101;
+                $self->log_request($reqstate);
+            },
+        );
+    };
+
+    return;
+}
+
+# Forward a request verbatim to a service on loopback, unlike proxy_request,
+# which re-encodes parsed parameters for another PVE node. The backend is a
+# foreign HTTP server, kept behind this server's TLS and authentication.
+sub local_http_proxy_request {
+    my ($self, $reqstate, $method, $target) = @_;
+
+    my $r = $reqstate->{request};
+
+    my $port = $target->{port};
+    my $socket = $target->{socket};
+    die "local_http_proxy_request: missing port or socket\n" if !$port && !$socket;
+    my $path = $target->{path} // '/';
+    my $scheme = $target->{tls} ? 'https' : 'http';
+
+    if ($r->header('upgrade')) {
+        $self->local_http_proxy_upgrade($reqstate, $method, $target);
+        return;
+    }
+
+    # Hop-by-hop headers describe the connection they arrived on, and
+    # Accept-Encoding goes too, so this server can compress the body itself.
+    my $skip = {
+        map { $_ => 1 } qw(
+            connection keep-alive host content-length transfer-encoding
+            upgrade te trailer proxy-authorization accept-encoding
+        )
+    };
+
+    # A unix socket has no authority to name, and nothing behind here routes on
+    # Host anyway.
+    my $headers = { Host => $port ? "127.0.0.1:$port" : 'localhost' };
+    $r->headers->scan(sub {
+        my ($key, $value) = @_;
+        $headers->{$key} = $value if !$skip->{ lc($key) };
+    });
+
+    my $content = $r->content;
+    $headers->{'Content-Length'} = length($content) if length($content);
+
+    my $tls_ctx;
+    if ($target->{tls}) {
+        # Loopback, with a certificate no browser sees and no CA signed: there
+        # is nothing verification could check.
+        $tls_ctx = AnyEvent::TLS->new(method => 'any', sslv2 => 0, sslv3 => 0, verify => 0);
+    }
+
+    # AnyEvent::HTTP needs a URL to parse, so a unix backend gets a nominal
+    # authority and a tcp_connect that ignores it.
+    my $url = $port ? "$scheme://127.0.0.1:$port$path" : "$scheme://localhost$path";
+    my $tcp_connect;
+    if ($socket) {
+        $tcp_connect = sub {
+            my (undef, undef, $connect_cb, $prepare_cb) = @_;
+            return AnyEvent::Socket::tcp_connect('unix/', $socket, $connect_cb, $prepare_cb);
+        };
+    }
+
+    my $w;
+    $w = http_request(
+        $method => $url,
+        headers => $headers,
+        $tcp_connect ? (tcp_connect => $tcp_connect) : (),
+        timeout => 30,
+        proxy => undef, # avoid use of $ENV{HTTP_PROXY}
+        persistent => 0,
+        keepalive => 0,
+        body => length($content) ? $content : undef,
+        $tls_ctx ? (tls_ctx => $tls_ctx) : (),
+        sub {
+            my ($body, $hdr) = @_;
+
+            undef $w;
+
+            if (!$reqstate->{hdl}) {
+                warn "local http proxy detected vanished client connection\n";
+                return;
+            }
+
+            eval {
+                my $code = delete $hdr->{Status};
+                my $msg = delete $hdr->{Reason};
+                delete $hdr->{URL};
+                delete $hdr->{HTTPVersion};
+
+                # AnyEvent::HTTP reports its own failures in the 59x range.
+                if ($code >= 590) {
+                    $self->error($reqstate, HTTP_BAD_GATEWAY, "$msg");
+                    return;
+                }
+
+                # Set by this server for the connection it answers on.
+                delete $hdr->{$_} for qw(connection transfer-encoding content-length);
+
+                my $header = HTTP::Headers->new(%$hdr);
+                my $resp = HTTP::Response->new($code, $msg, $header, $body);
+                # Note: disable compression, the backend decides its own encoding
+                $self->response($reqstate, $resp, undef, 1);
+            };
+            warn $@ if $@;
+        },
+    );
+
+    return;
+}
+
 sub proxy_request {
     my ($self, $reqstate, $clientip, $host, $node, $method, $uri, $auth, $params) = @_;
 
@@ -1222,6 +1466,25 @@ sub handle_request {
         # we re-enable timeout in response()
         $reqstate->{hdl}->timeout(0);
 
+        # The handler says where to send it, or nothing for the usual dispatch.
+        if (my $handler = $self->{local_http_proxy_handler}) {
+            my $target = eval { $handler->($self, $reqstate, $auth, $method, $path) };
+            if (my $err = $@) {
+                # The handler's refusals are answers: a denial must stay 403.
+                my $code = HTTP_INTERNAL_SERVER_ERROR;
+                if (ref($err) && eval { $err->{code} }) {
+                    my $carried = $err->{code};
+                    $code = $carried if $carried =~ m/^\d+$/ && $carried >= 400 && $carried <= 599;
+                }
+                $self->error($reqstate, $code, "$err");
+                return;
+            }
+            if ($target) {
+                $self->local_http_proxy_request($reqstate, $method, $target);
+                return;
+            }
+        }
+
         if ($path =~ m/^\Q$base_uri\E/) {
             $self->handle_api2_request($reqstate, $auth, $method, $path);
             return;
-- 
2.55.0