X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FHTTP.pm;h=b2c74cf378dcba2ab3f3cd3b6f9578ddb7feab58;hb=5c8909925072804901e9c3b45bbf25446d379e7b;hp=b8912950d1cc57af97da0c8ae24167ed9cb2d567;hpb=7c83d3e706811095cedab0bf62ac530d7b0f3a5a;p=public-inbox.git diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm index b8912950..b2c74cf3 100644 --- a/lib/PublicInbox/HTTP.pm +++ b/lib/PublicInbox/HTTP.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2016-2018 all contributors +# Copyright (C) 2016-2021 all contributors # License: AGPL-3.0+ # # Generic PSGI server for convenience. It aims to provide @@ -6,21 +6,30 @@ # to learn different ways to admin both NNTP and HTTP components. # There's nothing which depends on public-inbox, here. # Each instance of this class represents a HTTP client socket - +# +# fields: +# httpd: PublicInbox::HTTPD ref +# env: PSGI env hashref +# input_left: bytes left to read in request body (e.g. POST/PUT) +# remote_addr: remote IP address as a string (e.g. "127.0.0.1") +# remote_port: peer port +# forward: response body object, response to ->getline + ->close +# alive: HTTP keepalive state: +# 0: drop connection when done +# 1: keep connection when done +# 2: keep connection, chunk responses package PublicInbox::HTTP; use strict; -use warnings; -use base qw(PublicInbox::DS); -use fields qw(httpd env input_left remote_addr remote_port forward pull); -use bytes (); # only for bytes::length +use parent qw(PublicInbox::DS); use Fcntl qw(:seek); use Plack::HTTPParser qw(parse_http_request); # XS or pure Perl +use Plack::Util; use HTTP::Status qw(status_message); use HTTP::Date qw(time2str); -use IO::Handle; -require PublicInbox::EvCleanup; -PublicInbox::DS->import(qw(msg_more)); +use IO::Handle; # ->write +use PublicInbox::DS qw(msg_more); use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT); +use PublicInbox::Tmpfile; use constant { CHUNK_START => -1, # [a-f0-9]+\r\n CHUNK_END => -2, # \r\n @@ -55,12 +64,18 @@ sub http_date () { sub new ($$$) { my ($class, $sock, $addr, $httpd) = @_; - my $self = fields::new($class); - $self->SUPER::new($sock, EPOLLIN | EPOLLONESHOT); - $self->{httpd} = $httpd; + my $self = bless { httpd => $httpd }, $class; + my $ev = EPOLLIN; + my $wbuf; + if ($sock->can('accept_SSL') && !$sock->accept_SSL) { + return CORE::close($sock) if $! != EAGAIN; + $ev = PublicInbox::TLS::epollbit() or return CORE::close($sock); + $wbuf = [ \&PublicInbox::DS::accept_tls_step ]; + } + $self->{wbuf} = $wbuf if $wbuf; ($self->{remote_addr}, $self->{remote_port}) = PublicInbox::Daemon::host_with_port($addr); - $self; + $self->SUPER::new($sock, $ev | EPOLLONESHOT); } sub event_step { # called by PublicInbox::DS @@ -71,9 +86,9 @@ sub event_step { # called by PublicInbox::DS # only read more requests if we've drained the write buffer, # otherwise we can be buffering infinitely w/o backpressure - return read_input($self) if defined $self->{env}; + return read_input($self) if ref($self->{env}); my $rbuf = $self->{rbuf} // (\(my $x = '')); - $self->do_read($rbuf, 8192, bytes::length($$rbuf)) or return; + $self->do_read($rbuf, 8192, length($$rbuf)) or return; rbuf_process($self, $rbuf); } @@ -88,7 +103,7 @@ sub rbuf_process { # (they are rarely-used and git (as of 2.7.2) does not use them) if ($r == -1 || $env{HTTP_TRAILER} || # this length-check is necessary for PURE_PERL=1: - ($r == -2 && bytes::length($$rbuf) > 0x4000)) { + ($r == -2 && length($$rbuf) > 0x4000)) { return quit($self, 400); } if ($r < 0) { # incomplete @@ -105,7 +120,7 @@ sub rbuf_process { # IO::Handle::write returns boolean, this returns bytes written: sub xwrite ($$$) { my ($fh, $rbuf, $max) = @_; - my $w = bytes::length($$rbuf); + my $w = length($$rbuf); $w = $max if $w > $max; $fh->write($$rbuf, $w) or return; $w; @@ -115,7 +130,6 @@ sub read_input ($;$) { my ($self, $rbuf) = @_; $rbuf //= $self->{rbuf} // (\(my $x = '')); my $env = $self->{env}; - return if $env->{REMOTE_ADDR}; # in app dispatch return read_input_chunked($self, $rbuf) if env_chunked($env); # env->{CONTENT_LENGTH} (identity) @@ -144,9 +158,10 @@ sub app_dispatch { my ($self, $input, $rbuf) = @_; $self->rbuf_idle($rbuf); my $env = $self->{env}; + $self->{env} = undef; # for exists() check in ->busy $env->{REMOTE_ADDR} = $self->{remote_addr}; $env->{REMOTE_PORT} = $self->{remote_port}; - if (my $host = $env->{HTTP_HOST}) { + if (defined(my $host = $env->{HTTP_HOST})) { $host =~ s/:([0-9]+)\z// and $env->{SERVER_PORT} = $1; $env->{SERVER_NAME} = $host; } @@ -156,7 +171,7 @@ sub app_dispatch { } # note: NOT $self->{sock}, we want our close (+ PublicInbox::DS::close), # to do proper cleanup: - $env->{'psgix.io'} = $self; # only for ->close + $env->{'psgix.io'} = $self; # for ->close or async_pass my $res = Plack::Util::run_app($self->{httpd}->{app}, $env); eval { if (ref($res) eq 'CODE') { @@ -165,7 +180,10 @@ sub app_dispatch { response_write($self, $env, $res); } }; - $self->close if $@; + if ($@) { + err($self, "response_write error: $@"); + $self->close; + } } sub response_header_write { @@ -214,22 +232,20 @@ sub response_header_write { } # middlewares such as Deflater may write empty strings -sub chunked_wcb ($) { - my ($self) = @_; - sub { - return if $_[0] eq ''; - msg_more($self, sprintf("%x\r\n", bytes::length($_[0]))); - msg_more($self, $_[0]); - - # use $self->write(\"\n\n") if you care about real-time - # streaming responses, public-inbox WWW does not. - msg_more($self, "\r\n"); - } +sub chunked_write ($$) { + my $self = $_[0]; + return if $_[1] eq ''; + msg_more($self, sprintf("%x\r\n", length($_[1]))); + msg_more($self, $_[1]); + + # use $self->write(\"\n\n") if you care about real-time + # streaming responses, public-inbox WWW does not. + msg_more($self, "\r\n"); } -sub identity_wcb ($) { - my ($self) = @_; - sub { $self->write(\($_[0])) if $_[0] ne '' } +sub identity_write ($$) { + my $self = $_[0]; + $self->write(\($_[1])) if $_[1] ne ''; } sub next_request ($) { @@ -243,98 +259,113 @@ sub next_request ($) { } } -sub response_done_cb ($$) { +sub response_done { my ($self, $alive) = @_; - sub { - my $env = delete $self->{env}; - $self->write(\"0\r\n\r\n") if $alive == 2; - $self->write($alive ? \&next_request : \&close); - } + delete $self->{env}; # we're no longer busy + $self->write(\"0\r\n\r\n") if $alive == 2; + $self->write($alive ? \&next_request : \&close); } -sub getline_cb ($$$) { - my ($self, $write, $close) = @_; - local $/ = \8192; +sub getline_pull { + my ($self) = @_; my $forward = $self->{forward}; + # limit our own running time for fairness with other # clients and to avoid buffering too much: - if ($forward) { - my $buf = eval { $forward->getline }; - if (defined $buf) { - $write->($buf); # may close in PublicInbox::DS::write - if ($self->{sock}) { - my $next = $self->{pull}; - if ($self->{wbuf}) { - $self->write($next); - } else { - PublicInbox::DS::requeue($next); - } - return; - } - } elsif ($@) { - err($self, "response ->getline error: $@"); - $forward = undef; - $self->close; + my $buf = eval { + local $/ = \8192; + $forward->getline; + } if $forward; + + if (defined $buf) { + # may close in PublicInbox::DS::write + if ($self->{alive} == 2) { + chunked_write($self, $buf); + } else { + identity_write($self, $buf); } - } - delete @$self{qw(forward pull)}; + if ($self->{sock}) { + # autovivify wbuf + my $new_size = push(@{$self->{wbuf}}, \&getline_pull); + + # wbuf may be populated by {chunked,identity}_write() + # above, no need to rearm if so: + $self->requeue if $new_size == 1; + return; # likely + } + } elsif ($@) { + err($self, "response ->getline error: $@"); + $self->close; + } # avoid recursion - if ($forward) { + if (delete $self->{forward}) { eval { $forward->close }; if ($@) { err($self, "response ->close error: $@"); $self->close; # idempotent } } - $close->(); -} - -sub getline_response ($$$) { - my ($self, $write, $close) = @_; - my $pull = $self->{pull} = sub { getline_cb($self, $write, $close) }; - $pull->(); + response_done($self, delete $self->{alive}); } sub response_write { my ($self, $env, $res) = @_; my $alive = response_header_write($self, $env, $res); - my $close = response_done_cb($self, $alive); - my $write = $alive == 2 ? chunked_wcb($self) : identity_wcb($self); if (defined(my $body = $res->[2])) { if (ref $body eq 'ARRAY') { - $write->($_) foreach @$body; - $close->(); + if ($alive == 2) { + chunked_write($self, $_) for @$body; + } else { + identity_write($self, $_) for @$body; + } + response_done($self, $alive); } else { $self->{forward} = $body; - getline_response($self, $write, $close); + $self->{alive} = $alive; + getline_pull($self); # kick-off! } + # these are returned to the calling application: + } elsif ($alive == 2) { + bless [ $self, $alive ], 'PublicInbox::HTTP::Chunked'; } else { - # this is returned to the calling application: - Plack::Util::inline_object(write => $write, close => $close); + bless [ $self, $alive ], 'PublicInbox::HTTP::Identity'; } } sub input_tmpfile ($) { - open($_[0], '+>', undef); - $_[0]->autoflush(1); + my $input = tmpfile('http.input', $_[0]->{sock}) or return; + $input->autoflush(1); + $input; } sub input_prepare { my ($self, $env) = @_; - my $input; - my $len = $env->{CONTENT_LENGTH}; - if ($len) { - if ($len > $MAX_REQUEST_BUFFER) { - quit($self, 413); - return; - } - input_tmpfile($input); - } elsif (env_chunked($env)) { + my ($input, $len); + + # rfc 7230 3.3.2, 3.3.3,: favor Transfer-Encoding over Content-Length + my $hte = $env->{HTTP_TRANSFER_ENCODING}; + if (defined $hte) { + # rfc7230 3.3.3, point 3 says only chunked is accepted + # as the final encoding. Since neither public-inbox-httpd, + # git-http-backend, or our WWW-related code uses "gzip", + # "deflate" or "compress" as the Transfer-Encoding, we'll + # reject them: + return quit($self, 400) if $hte !~ /\Achunked\z/i; + $len = CHUNK_START; - input_tmpfile($input); + $input = input_tmpfile($self); } else { - $input = $null_io; + $len = $env->{CONTENT_LENGTH}; + if (defined $len) { + # rfc7230 3.3.3.4 + return quit($self, 400) if $len !~ /\A[0-9]+\z/; + + return quit($self, 413) if $len > $MAX_REQUEST_BUFFER; + $input = $len ? input_tmpfile($self) : $null_io; + } else { + $input = $null_io; + } } # TODO: expire idle clients on ENFILE / EMFILE @@ -345,7 +376,7 @@ sub input_prepare { $self->{input_left} = $len || 0; } -sub env_chunked { ($_[0]->{HTTP_TRANSFER_ENCODING} || '') =~ /\bchunked\b/i } +sub env_chunked { ($_[0]->{HTTP_TRANSFER_ENCODING} // '') =~ /\Achunked\z/i } sub err ($$) { eval { $_[0]->{httpd}->{env}->{'psgi.errors'}->print($_[1]."\n") }; @@ -379,12 +410,12 @@ sub read_input_chunked { # unlikely... $$rbuf =~ s/\A\r\n//s and return app_dispatch($self, $input, $rbuf); - return quit($self, 400) if bytes::length($$rbuf) > 2; + return quit($self, 400) if length($$rbuf) > 2; } if ($len == CHUNK_END) { if ($$rbuf =~ s/\A\r\n//s) { $len = CHUNK_START; - } elsif (bytes::length($$rbuf) > 2) { + } elsif (length($$rbuf) > 2) { return quit($self, 400); } } @@ -394,14 +425,14 @@ sub read_input_chunked { # unlikely... if (($len + -s $input) > $MAX_REQUEST_BUFFER) { return quit($self, 413); } - } elsif (bytes::length($$rbuf) > CHUNK_MAX_HDR) { + } elsif (length($$rbuf) > CHUNK_MAX_HDR) { return quit($self, 400); } # will break from loop since $len >= 0 } if ($len < 0) { # chunk header is trickled, read more - $self->do_read($rbuf, 8192, bytes::length($$rbuf)) or + $self->do_read($rbuf, 8192, length($$rbuf)) or return recv_err($self, $len); # (implicit) goto chunk_start if $r > 0; } @@ -438,14 +469,11 @@ sub quit { my $h = "HTTP/1.1 $status " . status_message($status) . "\r\n\r\n"; $self->write(\$h); $self->close; + undef; # input_prepare expects this } sub close { my $self = $_[0]; - if (my $env = delete $self->{env}) { - delete $env->{'psgix.io'}; # prevent circular references - } - delete $self->{pull}; if (my $forward = delete $self->{forward}) { eval { $forward->close }; err($self, "forward ->close error: $@") if $@; @@ -456,14 +484,39 @@ sub close { # for graceful shutdown in PublicInbox::Daemon: sub busy () { my ($self) = @_; - ($self->{rbuf} || $self->{env} || $self->{wbuf}); + ($self->{rbuf} || exists($self->{env}) || $self->{wbuf}); } -# fires after pending writes are complete: -sub restart_pass ($) { - $_[0]->{forward}->restart_read; # see PublicInbox::HTTPD::Async +# runs $cb on the next iteration of the event loop at earliest +sub next_step { + my ($self, $cb) = @_; + return unless exists $self->{sock}; + $self->requeue if 1 == push(@{$self->{wbuf}}, $cb); } -sub enqueue_restart_pass ($) { $_[0]->write(\&restart_pass) } +# Chunked and Identity packages are used for writing responses. +# They may be exposed to the PSGI application when the PSGI app +# returns a CODE ref for "push"-based responses +package PublicInbox::HTTP::Chunked; +use strict; + +sub write { + # ([$http], $buf) = @_; + PublicInbox::HTTP::chunked_write($_[0]->[0], $_[1]) +} + +sub close { + # $_[0] = [$http, $alive] + PublicInbox::HTTP::response_done(@{$_[0]}); +} + +package PublicInbox::HTTP::Identity; +use strict; +our @ISA = qw(PublicInbox::HTTP::Chunked); + +sub write { + # ([$http], $buf) = @_; + PublicInbox::HTTP::identity_write($_[0]->[0], $_[1]); +} 1;