[WIP v1] initial public-inbox-httpd implemenation

[WIP v1] initial public-inbox-httpd implemenation

[Eric Wong]

This is meant to provide an easy starting point for server admins.
It provides a basic HTTP server for admins unfamiliar with
configuring PSGI applications as well as being an identical
interface for management as our nntpd implementation.

This HTTP server may also be a generic Plack/PSGI server for
existing Plack/PSGI applications.
---
 lib/PublicInbox/HTTP.pm     | 305 ++++++++++++++++++++++++++++++++++++++++++++
 lib/PublicInbox/Listener.pm |   5 +-
 public-inbox-httpd          | 100 +++++++++++++++
 public-inbox-nntpd          |   2 +-
 t/httpd-corner.psgi         |  38 ++++++
 t/httpd-corner.t            | 114 +++++++++++++++++
 t/httpd.t                   | 116 +++++++++++++++++
 7 files changed, 677 insertions(+), 3 deletions(-)
 create mode 100644 lib/PublicInbox/HTTP.pm
 create mode 100644 public-inbox-httpd
 create mode 100644 t/httpd-corner.psgi
 create mode 100644 t/httpd-corner.t
 create mode 100644 t/httpd.t

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
new file mode 100644
index 0000000..6ea8ce6
--- /dev/null
+++ b/lib/PublicInbox/HTTP.pm
@@ -0,0 +1,305 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+#
+# Generic PSGI server for convenience.  It aims to provide
+# a consistent experience for public-inbox admins so they don't have
+# to learn different ways to admin both NNTP and HTTP components.
+# There's nothing public-inbox-specific, here.
+# Each instance of this class represents a HTTP client socket
+
+package PublicInbox::HTTP;
+use strict;
+use warnings;
+use base qw(Danga::Socket);
+use fields qw(httpd env rbuf input_left);
+use Fcntl qw(:seek);
+use HTTP::Parser::XS qw(parse_http_request); # supports pure Perl fallback
+use HTTP::Status qw(status_message);
+use HTTP::Date qw(time2str);
+use IO::File;
+my $null_io = IO::File->new('/dev/null', '<');
+my %CHUNK = (
+	START => -1,   # [a-f0-9]+\r\n
+	END => -2,     # \r\n
+	Z_START => -3, # 0\r\n
+	Z_END => -4    # \r\n
+);
+
+sub new ($$$) {
+	my ($class, $sock, $addr, $httpd) = @_;
+	my $self = fields::new($class);
+	$self->SUPER::new($sock);
+	$self->{httpd} = $httpd;
+	$self->watch_read(1);
+	$self;
+}
+
+sub event_read { # called by Danga::Socket
+	my ($self) = @_;
+
+	return event_read_input($self) if defined $self->{env};
+
+	my %env = %{$self->{httpd}->{env}}; # full hash copy
+	my $off = defined($self->{rbuf}) ? length($self->{rbuf}) : 0;
+	my $r = sysread($self->{sock}, $self->{rbuf}, 8192, $off);
+	unless (defined $r) {
+		return if $!{EAGAIN};
+		return $self->close;
+	}
+	return $self->close if $r == 0;
+
+	$r = parse_http_request($self->{rbuf}, \%env);
+	return $self->quit(400) if $r == -1;
+	return if $r < 0; # incomplete
+	$self->{rbuf} = substr($self->{rbuf}, $r);
+	my $len = input_prepare($self, \%env);
+	$len ? event_read_input($self) : app_dispatch($self);
+}
+
+sub event_read_input ($) {
+	my ($self) = @_;
+	my $env = $self->{env};
+	return event_read_input_chunked($self) if env_chunked($env);
+
+	# env->{CONTENT_LENGTH} (identity)
+	my $sock = $self->{sock};
+	my $len = $self->{input_left};
+	$self->{input_left} = undef;
+	my $rbuf = \($self->{rbuf});
+	my $input = $env->{'psgi.input'};
+
+	while ($len > 0) {
+		if ($$rbuf ne '') {
+			my $w = write_in_full($input, $rbuf, $len);
+			return $self->write_err unless $w;
+			$len -= $w;
+			die "BUG: $len < 0 (w=$w)" if $len < 0;
+			if ($len == 0) { # next request may be pipelined
+				$$rbuf = substr($$rbuf, $w);
+				last;
+			}
+			$$rbuf = '';
+		}
+		my $r = sysread($sock, $$rbuf, 8192);
+		return $self->recv_err($r, $len) unless $r;
+		# continue looping if $r > 0;
+	}
+	app_dispatch($self);
+}
+
+sub app_dispatch ($) {
+	my ($self) = @_;
+	$self->watch_read(0);
+	my $env = $self->{env};
+	$self->{env} = undef;
+	$env->{REMOTE_ADDR} = $self->peer_ip_string; # Danga::Socket
+	$env->{REMOTE_PORT} = $self->{peer_port}; # set by peer_ip_string
+	if (my $host = $env->{HTTP_HOST}) {
+		$env->{SERVER_NAME} = $env;
+	}
+	$env->{'psgi.input'}->seek(0, SEEK_SET);
+	my $res = Plack::Util::run_app($self->{httpd}->{app}, $env);
+	eval {
+		if (ref($res) eq 'CODE') {
+			$res->(sub { response_write($self, $env, $_[0]) });
+		} else {
+			response_write($self, $env, $res);
+		}
+	};
+	$self->close if $@;
+}
+
+sub response_header_write {
+	my ($self, $env, $res) = @_;
+	my $proto = $env->{SERVER_PROTOCOL} or return; # HTTP/0.9 :P
+	my $status = $res->[0];
+	my $h = "$proto $status " . status_message($status) . "\r\n";
+	my ($len, $chunked);
+	my $headers = $res->[1];
+
+	for (my $i = 0; $i < @$headers; $i += 2) {
+		my $k = $headers->[$i];
+		my $v = $headers->[$i + 1];
+		next if $k =~ /\A(?:Connection|Date)\z/i;
+
+		$len = $v if $k =~ /\AContent-Length\z/i;
+		if ($k =~ /\ATransfer-Encoding\z/i && $v =~ /\bchunked\b/i) {
+			$chunked = 1;
+		}
+
+		$h .= "$k: $v\r\n";
+	}
+
+	my $conn = $env->{HTTP_CONNECTION} || '';
+	my $alive = (defined($len) || $chunked) &&
+			($proto eq 'HTTP/1.1' && $conn !~ /\bclose\b/i) ||
+			($conn =~ /\bkeep-alive\b/i);
+
+	$h .= 'Connection: ' . ($alive ? 'keep-alive' : 'close');
+	$h .= "\r\nDate: " . time2str(time) . "\r\n\r\n";
+
+	if (($len || $chunked) && $env->{REQUEST_METHOD} ne 'HEAD') {
+		more($self, $h);
+	} else {
+		$self->write($h);
+	}
+	($alive, $chunked);
+}
+
+sub response_write {
+	my ($self, $env, $res) = @_;
+	my ($alive, $chunked) = response_header_write($self, $env, $res);
+	my $write = sub { $self->write($_[0]) };
+	my $close = sub {
+		if ($alive) {
+			$self->event_write; # watch for readability if done
+		} else {
+			$self->write(sub { $self->close });
+		}
+	};
+
+	if (defined $res->[2]) {
+		Plack::Util::foreach($res->[2], $write);
+		$close->();
+	} else {
+		# this is returned to the calling application:
+		Plack::Util::inline_object(write => $write, close => $close);
+	}
+}
+
+use constant MSG_MORE => ($^O eq 'linux') ? 0x8000 : 0;
+sub more ($$) {
+	my $self = $_[0];
+	if (MSG_MORE && !$self->{write_buf_size}) {
+		my $n = send($self->{sock}, $_[1], MSG_MORE);
+		if (defined $n) {
+			my $dlen = length($_[1]);
+			return 1 if $n == $dlen; # all done!
+			$_[1] = substr($_[1], $n, $dlen - $n);
+			# fall through to normal write:
+		}
+	}
+	$self->write($_[1]);
+}
+
+# overrides existing Danga::Socket method
+sub event_write {
+	my ($self) = @_;
+	# only continue watching for readability when we are done writing:
+	$self->watch_read(1) if $self->write(undef) == 1
+}
+
+sub input_prepare {
+	my ($self, $env) = @_;
+	my $input = $null_io;
+	my $len = $env->{CONTENT_LENGTH};
+	if ($len) {
+		$input = IO::File->new_tmpfile;
+	} elsif (env_chunked($env)) {
+		$input = IO::File->new_tmpfile;
+		$len = $CHUNK{START};
+	}
+	binmode $input;
+	$env->{'psgi.input'} = $input;
+	$self->{env} = $env;
+	$self->{input_left} = $len;
+}
+
+sub env_chunked { ($_[0]->{HTTP_TRANSFER_ENCODING} || '') =~ /\bchunked\b/i }
+
+sub write_err {
+	my ($self) = @_;
+	my $err = $self->{env}->{'psgi.errors'};
+	my $msg = $! || '(zero write)';
+	$err->print("error buffering to input: $msg\n");
+	$self->quit(500);
+}
+
+sub recv_err {
+	my ($self, $r, $len) = @_;
+	return $self->close if (defined $r && $r == 0);
+	if ($!{EAGAIN}) {
+		$self->{input_left} = $len;
+		return;
+	}
+	my $err = $self->{env}->{'psgi.errors'};
+	$err->print("error reading for input: $! ($len bytes remaining)\n");
+	$self->quit(500);
+}
+
+sub write_in_full {
+	my ($fh, $rbuf, $len) = @_;
+	my $rv = 0;
+	my $off = 0;
+	while ($len > 0) {
+		my $w = syswrite($fh, $$rbuf, $len, $off);
+		return ($rv ? $rv : $w) unless $w; # undef or 0
+		$rv += $w;
+		$off += $w;
+		$len -= $w;
+	}
+	$rv
+}
+
+sub event_read_input_chunked { # unlikely...
+	my ($self) = @_;
+	my $input = $self->{env}->{'psgi.input'};
+	my $sock = $self->{sock};
+	my $len = $self->{input_left};
+	$self->{input_left} = undef;
+	my $rbuf = \{$self->{rbuf}};
+
+	while ($len < 0) { # chunk start
+chunk_start:
+		if ($len == $CHUNK{Z_START} && $$rbuf =~ s/\r\n//) {
+			# final chunk seen
+			$len = $CHUNK{Z_END};
+			last;
+		}
+		if ($len == $CHUNK{END} && $$rbuf =~ s/\r\n//) {
+			$len = $CHUNK{START};
+		}
+		if ($len == $CHUNK{START} &&
+				$$rbuf =~ s/\A([a-f0-9]+).*?\r\n//i) {
+			$len = hex $1;
+			# will break from loop since $len >= 0
+		} else { # chunk header is trickled, read more
+			my $off = length($$rbuf);
+			my $r = sysread($sock, $$rbuf, 8192, $off);
+			return $self->recv_err($r, $len) unless $r;
+			# (implicit) goto chunk_start if $r > 0;
+		}
+	}
+	# drain the current chunk
+	until ($len == 0) {
+		my $w = write_in_full($input, $rbuf, $len);
+		return $self->write_err unless $w;
+		$len -= $w;
+		if ($len == 0) {
+			# we may have leftover data to parse in chunk
+			$$rbuf = substr($$rbuf, $w);
+			$len = $CHUNK{END};
+			goto chunk_start;
+		}
+		$$rbuf = '';
+
+		# read more of current chunk
+		my $r = sysread($sock, $$rbuf, 8192);
+		return $self->recv_err($r, $len) unless $r;
+		# continue looping if $r > 0;
+	}
+	if ($len == 0) { # final chunk starts with "0\r\n"
+		$len = $CHUNK{Z_START};
+		goto chunk_start;
+	}
+	app_dispatch($self);
+}
+
+sub quit {
+	my ($self, $status) = @_;
+	my $h = "HTTP/1.1 $status " . status_message($status) . "\r\n\r\n";
+	$self->write($h);
+	$self->close;
+}
+
+1;
diff --git a/lib/PublicInbox/Listener.pm b/lib/PublicInbox/Listener.pm
index 8e0554f..5f351a7 100644
--- a/lib/PublicInbox/Listener.pm
+++ b/lib/PublicInbox/Listener.pm
@@ -25,11 +25,12 @@ sub new ($$$) {
 
 sub event_read {
 	my ($self) = @_;
+	my $sock = $self->{sock};
 	# no loop here, we want to fairly distribute clients
 	# between multiple processes sharing the same socket
-	if (my $addr = accept(my $c, $self->{sock})) {
+	if (my $addr = accept(my $c, $sock)) {
 		IO::Handle::blocking($c, 0); # no accept4 :<
-		$self->{post_accept}->($c, $addr);
+		$self->{post_accept}->($c, $addr, $sock);
 	}
 }
 
diff --git a/public-inbox-httpd b/public-inbox-httpd
new file mode 100644
index 0000000..04b859c
--- /dev/null
+++ b/public-inbox-httpd
@@ -0,0 +1,100 @@
+#!/usr/bin/perl -w
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+#
+# Standalone HTTP server for public-inbox.
+use strict;
+use warnings;
+use Plack::Util;
+use PublicInbox::Daemon;
+use PublicInbox::HTTP;
+use PublicInbox::WWW;
+use Plack::Request;
+use Plack::Builder;
+PublicInbox::WWW->preload;
+my $have_deflater = eval { require Plack::Middleware::Deflater; 1 };
+my %httpds;
+my $config;
+my $app;
+my $refresh = sub {
+	if (@ARGV) {
+		eval { $app = Plack::Util::load_psgi(@ARGV) };
+		if ($@) {
+			die $@,
+"$0 runs in /, command-line paths must be absolute\n";
+		}
+	} else {
+		$app = eval {
+			my $deflate_types = eval {
+				require Plack::Middleware::Deflater;
+				[ 'text/html', 'text/plain',
+					'application/atom+xml' ]
+			};
+			builder {
+				enable 'Chunked';
+				if ($deflate_types) {
+					enable 'Deflater',
+						content_type => $deflate_types
+				}
+				enable 'Head';
+				sub {
+					my $req = Plack::Request->new(@_);
+					PublicInbox::WWW::run($req,
+							$req->method);
+				};
+			};
+		};
+	}
+};
+
+daemon_run('0.0.0.0:8080', $refresh,
+	sub ($$$) { # post_accept
+		my ($client, $addr, $srv) = @_;
+		my $fd = fileno($srv);
+		my $h = $httpds{$fd} ||= PublicInbox::HTTPD->new($srv, $app);
+		PublicInbox::HTTP->new($client, $addr, $h),
+	});
+
+1;
+package PublicInbox::HTTPD;
+use strict;
+use warnings;
+use Plack::Util;
+
+sub new {
+	my ($class, $sock, $app) = @_;
+	my $n = getsockname($sock) or die "not a socket: $sock $!\n";
+	my ($port, $addr);
+	if (length($n) >= 28) {
+		require Socket6;
+		($port, $addr) = Socket6::unpack_sockaddr_in6($n);
+	} else {
+		($port, $addr) = Socket::sockaddr_in($n);
+	}
+
+	my %env = (
+		REMOTE_HOST => '',
+		REMOTE_PORT => 0,
+		SERVER_NAME => $addr,
+		SERVER_PORT => $port,
+		SCRIPT_NAME => '',
+		'psgi.version' => [ 1, 1 ],
+		'psgi.errors' => \*STDERR,
+		'psgi.url_scheme' => 'http',
+		'psgi.nonblocking' => Plack::Util::TRUE,
+		'psgi.streaming' => Plack::Util::TRUE,
+		'psgi.run_once'	 => Plack::Util::FALSE,
+		'psgi.multithread' => Plack::Util::FALSE,
+		'psgi.multiprocess' => Plack::Util::TRUE,
+		'psgix.harakiri'=> Plack::Util::FALSE,
+		'psgix.input.buffered' => Plack::Util::TRUE,
+	);
+	bless {
+		err => \*STDERR,
+		out => \*STDOUT,
+		app => $app,
+		env => \%env,
+	}, $class;
+}
+
+1;
diff --git a/public-inbox-nntpd b/public-inbox-nntpd
index 706cbee..23d269d 100755
--- a/public-inbox-nntpd
+++ b/public-inbox-nntpd
@@ -12,7 +12,7 @@ require PublicInbox::Config;
 my $nntpd = PublicInbox::NNTPD->new;
 daemon_run('0.0.0.0:119',
 	sub { $nntpd->refresh_groups }, # refresh
-	sub ($$) { PublicInbox::NNTP->new($_[0], $nntpd) }); # post_accept
+	sub ($$$) { PublicInbox::NNTP->new($_[0], $nntpd) }); # post_accept
 
 1;
 package PublicInbox::NNTPD;
diff --git a/t/httpd-corner.psgi b/t/httpd-corner.psgi
new file mode 100644
index 0000000..3d9a73a
--- /dev/null
+++ b/t/httpd-corner.psgi
@@ -0,0 +1,38 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+# corner case tests for the generic PSGI server
+# Usage: plackup [OPTIONS] /path/to/this/file
+use strict;
+use warnings;
+use Plack::Request;
+use Plack::Builder;
+require Digest::SHA;
+my $app = sub {
+	my ($env) = @_;
+	my $path = $env->{PATH_INFO};
+	my $in = $env->{'psgi.input'};
+	my $actual = -s $in;
+	warn "len $env->{CONTENT_LENGTH} $actual\n";
+	my $code = 500;
+	my $h = [ 'Content-Type' => 'text/plain' ];
+	my $body = [];
+	if ($path eq '/sha1') {
+		my $sha1 = Digest::SHA->new('SHA-1');
+		my $buf;
+		while (1) {
+			my $r = $in->read($buf, 4096);
+			die "read err: $!" unless defined $r;
+			last if $r == 0;
+			$sha1->add($buf);
+		}
+		$code = 200;
+		push @$body, $sha1->hexdigest;
+	}
+	[ $code, $h, $body ]
+};
+
+builder {
+	enable 'ContentLength';
+	enable 'Head';
+	$app;
+}
diff --git a/t/httpd-corner.t b/t/httpd-corner.t
new file mode 100644
index 0000000..dd55669
--- /dev/null
+++ b/t/httpd-corner.t
@@ -0,0 +1,114 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+# note: our HTTP server should be standalone and capable of running
+# generic Rack apps.
+use strict;
+use warnings;
+use Test::More;
+
+foreach my $mod (qw(Plack::Util Plack::Request Plack::Builder Danga::Socket
+			HTTP::Parser::XS HTTP::Date HTTP::Status)) {
+	eval "require $mod";
+	plan skip_all => "$mod missing for httpd-corner.t" if $@;
+}
+
+use Digest::SHA qw(sha1_hex);
+use File::Temp qw/tempdir/;
+use Cwd qw/getcwd/;
+use IO::Socket;
+use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
+use Socket qw(SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
+my $tmpdir = tempdir(CLEANUP => 1);
+my $err = "$tmpdir/stderr.log";
+my $out = "$tmpdir/stdout.log";
+my $httpd = 'blib/script/public-inbox-httpd';
+my $psgi = getcwd()."/t/httpd-corner.psgi";
+my %opts = (
+	LocalAddr => '127.0.0.1',
+	ReuseAddr => 1,
+	Proto => 'tcp',
+	Type => SOCK_STREAM,
+	Listen => 1024,
+);
+my $sock = IO::Socket::INET->new(%opts);
+my $pid;
+END { kill 'TERM', $pid if defined $pid };
+{
+	ok($sock, 'sock created');
+	$! = 0;
+	my $fl = fcntl($sock, F_GETFD, 0);
+	ok(! $!, 'no error from fcntl(F_GETFD)');
+	is($fl, FD_CLOEXEC, 'cloexec set by default (Perl behavior)');
+	$pid = fork;
+	if ($pid == 0) {
+		use POSIX qw(dup2);
+		# pretend to be systemd
+		fcntl($sock, F_SETFD, $fl &= ~FD_CLOEXEC);
+		dup2(fileno($sock), 3) or die "dup2 failed: $!\n";
+		$ENV{LISTEN_PID} = $$;
+		$ENV{LISTEN_FDS} = 1;
+		exec $httpd, '-W0', "--stdout=$out", "--stderr=$err", $psgi;
+		die "FAIL: $!\n";
+	}
+	ok(defined $pid, 'forked httpd process successfully');
+	$! = 0;
+	fcntl($sock, F_SETFD, $fl |= FD_CLOEXEC);
+	ok(! $!, 'no error from fcntl(F_SETFD)');
+}
+
+sub conn_for {
+	my ($sock, $msg) = @_;
+	my $conn = IO::Socket::INET->new(
+				PeerAddr => $sock->sockhost,
+				PeerPort => $sock->sockport,
+				Proto => 'tcp',
+				Type => SOCK_STREAM);
+	ok($conn, "connected for $msg");
+	return $conn;
+}
+
+sub delay { select(undef, undef, undef, 0.2) }
+
+my $str = 'abcdefghijklmnopqrstuvwxyz';
+my $len = length $str;
+is($len, 26, 'got the alphabet');
+my $check_self = sub {
+	my ($conn) = @_;
+	$conn->read(my $buf, 4096);
+	my ($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr/\r\nContent-Length: 40\r\n/s, 'got expected length');
+	is($body, sha1_hex($str), 'read expected body');
+};
+
+{
+	my $conn = conn_for($sock, 'trickle body');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n");
+	$conn->write("Content-Length: $len\r\n\r\n");
+	my $beg = substr($str, 0, 10);
+	my $end = substr($str, 10);
+	is($beg . $end, $str, 'substr setup correct');
+	delay();
+	$conn->write($beg);
+	delay();
+	$conn->write($end);
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'one-shot write');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n" .
+			"Content-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'trickle header, one-shot body');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n");
+	delay();
+	$conn->write("Content-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+done_testing();
+
+1;
diff --git a/t/httpd.t b/t/httpd.t
new file mode 100644
index 0000000..034f1ae
--- /dev/null
+++ b/t/httpd.t
@@ -0,0 +1,116 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+use strict;
+use warnings;
+use Test::More;
+
+foreach my $mod (qw(Plack::Util Plack::Request Plack::Builder Danga::Socket
+			HTTP::Parser::XS HTTP::Date HTTP::Status)) {
+	eval "require $mod";
+	plan skip_all => "$mod missing for httpd.t" if $@;
+}
+use File::Temp qw/tempdir/;
+use Cwd qw/getcwd/;
+use IO::Socket;
+use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
+use Socket qw(SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
+use IPC::Run;
+
+# FIXME: too much setup
+my $tmpdir = tempdir(CLEANUP => 1);
+my $home = "$tmpdir/pi-home";
+my $err = "$tmpdir/stderr.log";
+my $out = "$tmpdir/stdout.log";
+my $pi_home = "$home/.public-inbox";
+my $pi_config = "$pi_home/config";
+my $maindir = "$tmpdir/main.git";
+my $main_bin = getcwd()."/t/main-bin";
+my $main_path = "$main_bin:$ENV{PATH}"; # for spamc ham mock
+my $group = 'test-httpd';
+my $addr = $group . '@example.com';
+my $cfgpfx = "publicinbox.$group";
+my $failbox = "$home/fail.mbox";
+local $ENV{PI_EMERGENCY} = $failbox;
+my $mda = 'blib/script/public-inbox-mda';
+my $httpd = 'blib/script/public-inbox-httpd';
+my $init = 'blib/script/public-inbox-init';
+
+my %opts = (
+	LocalAddr => '127.0.0.1',
+	ReuseAddr => 1,
+	Proto => 'tcp',
+	Type => SOCK_STREAM,
+	Listen => 1024,
+);
+my $sock = IO::Socket::INET->new(%opts);
+my $pid;
+END { kill 'TERM', $pid if defined $pid };
+{
+	local $ENV{HOME} = $home;
+	ok(!system($init, $group, $maindir, 'http://example.com/', $addr),
+		'init ran properly');
+
+	# ensure successful message delivery
+	{
+		local $ENV{ORIGINAL_RECIPIENT} = $addr;
+		my $in = <<EOF;
+From: Me <me\@example.com>
+To: You <you\@example.com>
+Cc: $addr
+Message-Id: <nntp\@example.com>
+Subject: hihi
+Date: Thu, 01 Jan 1970 06:06:06 +0000
+
+nntp
+EOF
+		local $ENV{PATH} = $main_path;
+		IPC::Run::run([$mda], \$in);
+		is(0, $?, 'ran MDA correctly');
+	}
+	ok($sock, 'sock created');
+	$! = 0;
+	my $fl = fcntl($sock, F_GETFD, 0);
+	ok(! $!, 'no error from fcntl(F_GETFD)');
+	is($fl, FD_CLOEXEC, 'cloexec set by default (Perl behavior)');
+	$pid = fork;
+	if ($pid == 0) {
+		use POSIX qw(dup2);
+		# pretend to be systemd
+		fcntl($sock, F_SETFD, $fl &= ~FD_CLOEXEC);
+		dup2(fileno($sock), 3) or die "dup2 failed: $!\n";
+		$ENV{LISTEN_PID} = $$;
+		$ENV{LISTEN_FDS} = 1;
+		exec $httpd, "--stdout=$out", "--stderr=$err";
+		die "FAIL: $!\n";
+	}
+	ok(defined $pid, 'forked httpd process successfully');
+	$! = 0;
+	fcntl($sock, F_SETFD, $fl |= FD_CLOEXEC);
+	ok(! $!, 'no error from fcntl(F_SETFD)');
+	my $host = $sock->sockhost;
+	my $port = $sock->sockport;
+	my $conn = IO::Socket::INET->new(PeerAddr => $host,
+				PeerPort => $port,
+				Proto => 'tcp',
+				Type => SOCK_STREAM);
+	ok($conn, 'connected');
+	ok($conn->write("GET / HTTP/1.0\r\n\r\n"), 'wrote data to socket');
+	{
+		my $buf;
+		ok($conn->read($buf, 4096), 'read some bytes');
+		like($buf, qr!\AHTTP/1\.[01] 404\b!, 'got 404 response');
+		is($conn->read($buf, 1), 0, "EOF");
+	}
+
+	is(system(qw(git clone -q --mirror),
+			"http://$host:$port/$group", "$tmpdir/clone.git"),
+		0, 'clone successful');
+
+	ok(kill('TERM', $pid), 'killed httpd');
+	$pid = undef;
+	waitpid(-1, 0);
+}
+
+done_testing();
+
+1;
-- 
EW

[PATCH] initial public-inbox-httpd implemenation

[Eric Wong]

This is meant to provide an easy starting point for server admins.
It provides a basic HTTP server for admins unfamiliar with
configuring PSGI applications as well as being an identical
interface for management as our nntpd implementation.

This HTTP server may also be a generic Plack/PSGI server for
existing Plack/PSGI applications.
---
 (pushed and running on http://public-inbox.org/meta/ !)

 lib/PublicInbox/HTTP.pm     | 334 ++++++++++++++++++++++++++++++++++++++++++++
 lib/PublicInbox/Listener.pm |   5 +-
 public-inbox-httpd          | 100 +++++++++++++
 public-inbox-nntpd          |   2 +-
 t/httpd-corner.psgi         |  37 +++++
 t/httpd-corner.t            | 286 +++++++++++++++++++++++++++++++++++++
 t/httpd.t                   | 119 ++++++++++++++++
 7 files changed, 880 insertions(+), 3 deletions(-)
 create mode 100644 lib/PublicInbox/HTTP.pm
 create mode 100644 public-inbox-httpd
 create mode 100644 t/httpd-corner.psgi
 create mode 100644 t/httpd-corner.t
 create mode 100644 t/httpd.t

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
new file mode 100644
index 0000000..a5149ac
--- /dev/null
+++ b/lib/PublicInbox/HTTP.pm
@@ -0,0 +1,334 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+#
+# Generic PSGI server for convenience.  It aims to provide
+# a consistent experience for public-inbox admins so they don't have
+# to learn different ways to admin both NNTP and HTTP components.
+# There's nothing public-inbox-specific, here.
+# Each instance of this class represents a HTTP client socket
+
+package PublicInbox::HTTP;
+use strict;
+use warnings;
+use base qw(Danga::Socket);
+use fields qw(httpd env rbuf input_left);
+use Fcntl qw(:seek);
+use HTTP::Parser::XS qw(parse_http_request); # supports pure Perl fallback
+use HTTP::Status qw(status_message);
+use HTTP::Date qw(time2str);
+use IO::File;
+my $null_io = IO::File->new('/dev/null', '<');
+use constant {
+	CHUNK_START => -1,   # [a-f0-9]+\r\n
+	CHUNK_END => -2,     # \r\n
+	CHUNK_ZEND => -3,    # \r\n
+	CHUNK_MAX_HDR => 256,
+};
+
+sub new ($$$) {
+	my ($class, $sock, $addr, $httpd) = @_;
+	my $self = fields::new($class);
+	$self->SUPER::new($sock);
+	$self->{httpd} = $httpd;
+	$self->{rbuf} = '';
+	$self->watch_read(1);
+	$self;
+}
+
+sub event_read { # called by Danga::Socket
+	my ($self) = @_;
+
+	return event_read_input($self) if defined $self->{env};
+
+	my $off = $self->{rbuf} eq '' ? 0 : length($self->{rbuf});
+	my $r = sysread($self->{sock}, $self->{rbuf}, 8192, $off);
+	if (defined $r) {
+		return $self->close if $r == 0;
+		return rbuf_process($self);
+	}
+	return if $!{EAGAIN}; # no need to call watch_read(1) again
+
+	# common for clients to break connections without warning,
+	# would be too noisy to log here:
+	return $self->close;
+}
+
+sub rbuf_process {
+	my ($self) = @_;
+
+	my %env = %{$self->{httpd}->{env}}; # full hash copy
+	my $r = parse_http_request($self->{rbuf}, \%env);
+
+	# We do not support Trailers in chunked requests, for now
+	# (they are rarely-used and git (as of 2.7.2) does not use them)
+	return $self->quit(400) if $r == -1 || $env{HTTP_TRAILER};
+	return $self->watch_read(1) if $r < 0; # incomplete
+	$self->{rbuf} = substr($self->{rbuf}, $r);
+	my $len = input_prepare($self, \%env);
+	$len ? event_read_input($self) : app_dispatch($self);
+}
+
+sub event_read_input ($) {
+	my ($self) = @_;
+	my $env = $self->{env};
+	return event_read_input_chunked($self) if env_chunked($env);
+
+	# env->{CONTENT_LENGTH} (identity)
+	my $sock = $self->{sock};
+	my $len = $self->{input_left};
+	$self->{input_left} = undef;
+	my $rbuf = \($self->{rbuf});
+	my $input = $env->{'psgi.input'};
+
+	while ($len > 0) {
+		if ($$rbuf ne '') {
+			my $w = write_in_full($input, $rbuf, $len);
+			return $self->write_err unless $w;
+			$len -= $w;
+			die "BUG: $len < 0 (w=$w)" if $len < 0;
+			if ($len == 0) { # next request may be pipelined
+				$$rbuf = substr($$rbuf, $w);
+				last;
+			}
+			$$rbuf = '';
+		}
+		my $r = sysread($sock, $$rbuf, 8192);
+		return $self->recv_err($r, $len) unless $r;
+		# continue looping if $r > 0;
+	}
+	app_dispatch($self);
+}
+
+sub app_dispatch ($) {
+	my ($self) = @_;
+	$self->watch_read(0);
+	my $env = $self->{env};
+	$self->{env} = undef;
+	$env->{REMOTE_ADDR} = $self->peer_ip_string; # Danga::Socket
+	$env->{REMOTE_PORT} = $self->{peer_port}; # set by peer_ip_string
+	if (my $host = $env->{HTTP_HOST}) {
+		$host =~ s/:(\d+)\z// and $env->{SERVER_PORT} = $1;
+		$env->{SERVER_NAME} = $host;
+	}
+	$env->{'psgi.input'}->seek(0, SEEK_SET);
+	my $res = Plack::Util::run_app($self->{httpd}->{app}, $env);
+	eval {
+		if (ref($res) eq 'CODE') {
+			$res->(sub { response_write($self, $env, $_[0]) });
+		} else {
+			response_write($self, $env, $res);
+		}
+	};
+	$self->close if $@;
+}
+
+sub response_header_write {
+	my ($self, $env, $res) = @_;
+	my $proto = $env->{SERVER_PROTOCOL} or return; # HTTP/0.9 :P
+	my $status = $res->[0];
+	my $h = "$proto $status " . status_message($status) . "\r\n";
+	my ($len, $chunked);
+	my $headers = $res->[1];
+
+	for (my $i = 0; $i < @$headers; $i += 2) {
+		my $k = $headers->[$i];
+		my $v = $headers->[$i + 1];
+		next if $k =~ /\A(?:Connection|Date)\z/i;
+
+		$len = $v if $k =~ /\AContent-Length\z/i;
+		if ($k =~ /\ATransfer-Encoding\z/i && $v =~ /\bchunked\b/i) {
+			$chunked = 1;
+		}
+
+		$h .= "$k: $v\r\n";
+	}
+
+	my $conn = $env->{HTTP_CONNECTION} || '';
+	my $alive = (defined($len) || $chunked) &&
+			($proto eq 'HTTP/1.1' && $conn !~ /\bclose\b/i) ||
+			($conn =~ /\bkeep-alive\b/i);
+
+	$h .= 'Connection: ' . ($alive ? 'keep-alive' : 'close');
+	$h .= "\r\nDate: " . time2str(time) . "\r\n\r\n";
+
+	if (($len || $chunked) && $env->{REQUEST_METHOD} ne 'HEAD') {
+		more($self, $h);
+	} else {
+		$self->write($h);
+	}
+	($alive, $chunked);
+}
+
+sub response_write {
+	my ($self, $env, $res) = @_;
+	my ($alive, $chunked) = response_header_write($self, $env, $res);
+	my $write = sub { $self->write($_[0]) };
+	my $close = sub {
+		if ($alive) {
+			$self->event_write; # watch for readability if done
+		} else {
+			$self->write(sub { $self->close });
+		}
+	};
+
+	if (defined $res->[2]) {
+		Plack::Util::foreach($res->[2], $write);
+		$close->();
+	} else {
+		# this is returned to the calling application:
+		Plack::Util::inline_object(write => $write, close => $close);
+	}
+}
+
+use constant MSG_MORE => ($^O eq 'linux') ? 0x8000 : 0;
+sub more ($$) {
+	my $self = $_[0];
+	if (MSG_MORE && !$self->{write_buf_size}) {
+		my $n = send($self->{sock}, $_[1], MSG_MORE);
+		if (defined $n) {
+			my $dlen = length($_[1]);
+			return 1 if $n == $dlen; # all done!
+			$_[1] = substr($_[1], $n, $dlen - $n);
+			# fall through to normal write:
+		}
+	}
+	$self->write($_[1]);
+}
+
+# overrides existing Danga::Socket method
+sub event_write {
+	my ($self) = @_;
+	# only continue watching for readability when we are done writing:
+	return if $self->write(undef) != 1;
+
+	if ($self->{rbuf} eq '') {
+		$self->watch_read(1);
+	} else {
+		# avoid recursion
+		Danga::Socket->AddTimer(0, sub { rbuf_process($self) });
+	}
+}
+
+sub input_prepare {
+	my ($self, $env) = @_;
+	my $input = $null_io;
+	my $len = $env->{CONTENT_LENGTH};
+	if ($len) {
+		$input = IO::File->new_tmpfile;
+	} elsif (env_chunked($env)) {
+		$input = IO::File->new_tmpfile;
+		$len = CHUNK_START;
+	}
+	binmode $input;
+	$env->{'psgi.input'} = $input;
+	$self->{env} = $env;
+	$self->{input_left} = $len;
+}
+
+sub env_chunked { ($_[0]->{HTTP_TRANSFER_ENCODING} || '') =~ /\bchunked\b/i }
+
+sub write_err {
+	my ($self) = @_;
+	my $err = $self->{env}->{'psgi.errors'};
+	my $msg = $! || '(zero write)';
+	$err->print("error buffering to input: $msg\n");
+	$self->quit(500);
+}
+
+sub recv_err {
+	my ($self, $r, $len) = @_;
+	return $self->close if (defined $r && $r == 0);
+	if ($!{EAGAIN}) {
+		$self->{input_left} = $len;
+		return;
+	}
+	my $err = $self->{env}->{'psgi.errors'};
+	$err->print("error reading for input: $! ($len bytes remaining)\n");
+	$self->quit(500);
+}
+
+sub write_in_full {
+	my ($fh, $rbuf, $len) = @_;
+	my $rv = 0;
+	my $off = 0;
+	while ($len > 0) {
+		my $w = syswrite($fh, $$rbuf, $len, $off);
+		return ($rv ? $rv : $w) unless $w; # undef or 0
+		$rv += $w;
+		$off += $w;
+		$len -= $w;
+	}
+	$rv
+}
+
+sub event_read_input_chunked { # unlikely...
+	my ($self) = @_;
+	my $input = $self->{env}->{'psgi.input'};
+	my $sock = $self->{sock};
+	my $len = $self->{input_left};
+	$self->{input_left} = undef;
+	my $rbuf = \($self->{rbuf});
+
+	while (1) { # chunk start
+		if ($len == CHUNK_ZEND) {
+			return app_dispatch($self) if $$rbuf =~ s/\A\r\n//s;
+			return $self->quit(400) if length($$rbuf) > 2;
+		}
+		if ($len == CHUNK_END) {
+			if ($$rbuf =~ s/\A\r\n//s) {
+				$len = CHUNK_START;
+			} elsif (length($$rbuf) > 2) {
+				return $self->quit(400);
+			}
+		}
+		if ($len == CHUNK_START) {
+			if ($$rbuf =~ s/\A([a-f0-9]+).*?\r\n//i) {
+				$len = hex $1;
+			} elsif (length($$rbuf) > CHUNK_MAX_HDR) {
+				return $self->quit(400);
+			}
+			# will break from loop since $len >= 0
+		}
+
+		if ($len < 0) { # chunk header is trickled, read more
+			my $off = length($$rbuf);
+			my $r = sysread($sock, $$rbuf, 8192, $off);
+			return $self->recv_err($r, $len) unless $r;
+			# (implicit) goto chunk_start if $r > 0;
+		}
+		$len = CHUNK_ZEND if $len == 0;
+
+		# drain the current chunk
+		until ($len <= 0) {
+			if ($$rbuf ne '') {
+				my $w = write_in_full($input, $rbuf, $len);
+				return $self->write_err unless $w;
+				$len -= $w;
+				if ($len == 0) {
+					# we may have leftover data to parse
+					# in chunk
+					$$rbuf = substr($$rbuf, $w);
+					$len = CHUNK_END;
+				} elsif ($len < 0) {
+					die "BUG: len < 0: $len";
+				} else {
+					$$rbuf = '';
+				}
+			}
+			if ($$rbuf eq '') {
+				# read more of current chunk
+				my $r = sysread($sock, $$rbuf, 8192);
+				return $self->recv_err($r, $len) unless $r;
+			}
+		}
+	}
+}
+
+sub quit {
+	my ($self, $status) = @_;
+	my $h = "HTTP/1.1 $status " . status_message($status) . "\r\n\r\n";
+	$self->write($h);
+	$self->close;
+}
+
+1;
diff --git a/lib/PublicInbox/Listener.pm b/lib/PublicInbox/Listener.pm
index 8e0554f..5f351a7 100644
--- a/lib/PublicInbox/Listener.pm
+++ b/lib/PublicInbox/Listener.pm
@@ -25,11 +25,12 @@ sub new ($$$) {
 
 sub event_read {
 	my ($self) = @_;
+	my $sock = $self->{sock};
 	# no loop here, we want to fairly distribute clients
 	# between multiple processes sharing the same socket
-	if (my $addr = accept(my $c, $self->{sock})) {
+	if (my $addr = accept(my $c, $sock)) {
 		IO::Handle::blocking($c, 0); # no accept4 :<
-		$self->{post_accept}->($c, $addr);
+		$self->{post_accept}->($c, $addr, $sock);
 	}
 }
 
diff --git a/public-inbox-httpd b/public-inbox-httpd
new file mode 100644
index 0000000..6436bd7
--- /dev/null
+++ b/public-inbox-httpd
@@ -0,0 +1,100 @@
+#!/usr/bin/perl -w
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+#
+# Standalone HTTP server for public-inbox.
+use strict;
+use warnings;
+use Plack::Util;
+use PublicInbox::Daemon;
+use PublicInbox::HTTP;
+use PublicInbox::WWW;
+use Plack::Request;
+use Plack::Builder;
+PublicInbox::WWW->preload;
+my $have_deflater = eval { require Plack::Middleware::Deflater; 1 };
+my %httpds;
+my $config;
+my $app;
+my $refresh = sub {
+	if (@ARGV) {
+		eval { $app = Plack::Util::load_psgi(@ARGV) };
+		if ($@) {
+			die $@,
+"$0 runs in /, command-line paths must be absolute\n";
+		}
+	} else {
+		$app = eval {
+			my $deflate_types = eval {
+				require Plack::Middleware::Deflater;
+				[ 'text/html', 'text/plain',
+					'application/atom+xml' ]
+			};
+			builder {
+				enable 'Chunked';
+				if ($deflate_types) {
+					enable 'Deflater',
+						content_type => $deflate_types
+				}
+				enable 'Head';
+				sub {
+					my $req = Plack::Request->new(@_);
+					PublicInbox::WWW::run($req,
+							$req->method);
+				};
+			};
+		};
+	}
+};
+
+daemon_run('0.0.0.0:8080', $refresh,
+	sub ($$$) { # post_accept
+		my ($client, $addr, $srv) = @_;
+		my $fd = fileno($srv);
+		my $h = $httpds{$fd} ||= PublicInbox::HTTPD->new($srv, $app);
+		PublicInbox::HTTP->new($client, $addr, $h),
+	});
+
+1;
+package PublicInbox::HTTPD;
+use strict;
+use warnings;
+use Plack::Util;
+
+sub new {
+	my ($class, $sock, $app) = @_;
+	my $n = getsockname($sock) or die "not a socket: $sock $!\n";
+	my ($port, $addr);
+	if (length($n) >= 28) {
+		require Socket6;
+		($port, $addr) = Socket6::unpack_sockaddr_in6($n);
+	} else {
+		($port, $addr) = Socket::unpack_sockaddr_in($n);
+	}
+
+	my %env = (
+		REMOTE_HOST => '',
+		REMOTE_PORT => 0,
+		SERVER_NAME => $addr,
+		SERVER_PORT => $port,
+		SCRIPT_NAME => '',
+		'psgi.version' => [ 1, 1 ],
+		'psgi.errors' => \*STDERR,
+		'psgi.url_scheme' => 'http',
+		'psgi.nonblocking' => Plack::Util::TRUE,
+		'psgi.streaming' => Plack::Util::TRUE,
+		'psgi.run_once'	 => Plack::Util::FALSE,
+		'psgi.multithread' => Plack::Util::FALSE,
+		'psgi.multiprocess' => Plack::Util::TRUE,
+		'psgix.harakiri'=> Plack::Util::FALSE,
+		'psgix.input.buffered' => Plack::Util::TRUE,
+	);
+	bless {
+		err => \*STDERR,
+		out => \*STDOUT,
+		app => $app,
+		env => \%env,
+	}, $class;
+}
+
+1;
diff --git a/public-inbox-nntpd b/public-inbox-nntpd
index 706cbee..23d269d 100755
--- a/public-inbox-nntpd
+++ b/public-inbox-nntpd
@@ -12,7 +12,7 @@ require PublicInbox::Config;
 my $nntpd = PublicInbox::NNTPD->new;
 daemon_run('0.0.0.0:119',
 	sub { $nntpd->refresh_groups }, # refresh
-	sub ($$) { PublicInbox::NNTP->new($_[0], $nntpd) }); # post_accept
+	sub ($$$) { PublicInbox::NNTP->new($_[0], $nntpd) }); # post_accept
 
 1;
 package PublicInbox::NNTPD;
diff --git a/t/httpd-corner.psgi b/t/httpd-corner.psgi
new file mode 100644
index 0000000..1947f37
--- /dev/null
+++ b/t/httpd-corner.psgi
@@ -0,0 +1,37 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+# corner case tests for the generic PSGI server
+# Usage: plackup [OPTIONS] /path/to/this/file
+use strict;
+use warnings;
+use Plack::Request;
+use Plack::Builder;
+require Digest::SHA;
+my $app = sub {
+	my ($env) = @_;
+	my $path = $env->{PATH_INFO};
+	my $in = $env->{'psgi.input'};
+	my $actual = -s $in;
+	my $code = 500;
+	my $h = [ 'Content-Type' => 'text/plain' ];
+	my $body = [];
+	if ($path eq '/sha1') {
+		my $sha1 = Digest::SHA->new('SHA-1');
+		my $buf;
+		while (1) {
+			my $r = $in->read($buf, 4096);
+			die "read err: $!" unless defined $r;
+			last if $r == 0;
+			$sha1->add($buf);
+		}
+		$code = 200;
+		push @$body, $sha1->hexdigest;
+	}
+	[ $code, $h, $body ]
+};
+
+builder {
+	enable 'ContentLength';
+	enable 'Head';
+	$app;
+}
diff --git a/t/httpd-corner.t b/t/httpd-corner.t
new file mode 100644
index 0000000..5834c1b
--- /dev/null
+++ b/t/httpd-corner.t
@@ -0,0 +1,286 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+# note: our HTTP server should be standalone and capable of running
+# generic Rack apps.
+use strict;
+use warnings;
+use Test::More;
+
+foreach my $mod (qw(Plack::Util Plack::Request Plack::Builder Danga::Socket
+			HTTP::Parser::XS HTTP::Date HTTP::Status)) {
+	eval "require $mod";
+	plan skip_all => "$mod missing for httpd-corner.t" if $@;
+}
+
+use Digest::SHA qw(sha1_hex);
+use File::Temp qw/tempdir/;
+use Cwd qw/getcwd/;
+use IO::Socket;
+use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
+use Socket qw(SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
+my $tmpdir = tempdir(CLEANUP => 1);
+my $err = "$tmpdir/stderr.log";
+my $out = "$tmpdir/stdout.log";
+my $httpd = 'blib/script/public-inbox-httpd';
+my $psgi = getcwd()."/t/httpd-corner.psgi";
+my %opts = (
+	LocalAddr => '127.0.0.1',
+	ReuseAddr => 1,
+	Proto => 'tcp',
+	Type => SOCK_STREAM,
+	Listen => 1024,
+);
+my $sock = IO::Socket::INET->new(%opts);
+my $pid;
+END { kill 'TERM', $pid if defined $pid };
+{
+	ok($sock, 'sock created');
+	$! = 0;
+	my $fl = fcntl($sock, F_GETFD, 0);
+	ok(! $!, 'no error from fcntl(F_GETFD)');
+	is($fl, FD_CLOEXEC, 'cloexec set by default (Perl behavior)');
+	$pid = fork;
+	if ($pid == 0) {
+		use POSIX qw(dup2);
+		# pretend to be systemd
+		fcntl($sock, F_SETFD, $fl &= ~FD_CLOEXEC);
+		dup2(fileno($sock), 3) or die "dup2 failed: $!\n";
+		$ENV{LISTEN_PID} = $$;
+		$ENV{LISTEN_FDS} = 1;
+		exec $httpd, '-W0', "--stdout=$out", "--stderr=$err", $psgi;
+		die "FAIL: $!\n";
+	}
+	ok(defined $pid, 'forked httpd process successfully');
+	$! = 0;
+	fcntl($sock, F_SETFD, $fl |= FD_CLOEXEC);
+	ok(! $!, 'no error from fcntl(F_SETFD)');
+}
+
+sub conn_for {
+	my ($sock, $msg) = @_;
+	my $conn = IO::Socket::INET->new(
+				PeerAddr => $sock->sockhost,
+				PeerPort => $sock->sockport,
+				Proto => 'tcp',
+				Type => SOCK_STREAM);
+	ok($conn, "connected for $msg");
+	$conn->autoflush(1);
+	setsockopt($conn, IPPROTO_TCP, TCP_NODELAY, 1);
+	return $conn;
+}
+
+sub delay { select(undef, undef, undef, shift || rand(0.02)) }
+
+my $str = 'abcdefghijklmnopqrstuvwxyz';
+my $len = length $str;
+is($len, 26, 'got the alphabet');
+my $check_self = sub {
+	my ($conn) = @_;
+	$conn->read(my $buf, 4096);
+	my ($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr/\r\nContent-Length: 40\r\n/s, 'got expected length');
+	is($body, sha1_hex($str), 'read expected body');
+};
+
+{
+	my $conn = conn_for($sock, '1.1 pipeline together');
+	$conn->write("PUT /sha1 HTTP/1.1\r\nUser-agent: hello\r\n\r\n" .
+			"PUT /sha1 HTTP/1.1\r\n\r\n");
+	my $buf = '';
+	my @r;
+	until (scalar(@r) >= 2) {
+		my $r = $conn->sysread(my $tmp, 4096);
+		die $! unless defined $r;
+		die "EOF <$buf>" unless $r;
+		$buf .= $tmp;
+		@r = ($buf =~ /\r\n\r\n([a-f0-9]{40})/g);
+	}
+	is(2, scalar @r, 'got 2 responses');
+	my $i = 3;
+	foreach my $hex (@r) {
+		is($hex, sha1_hex(''), "read expected body $i");
+		$i++;
+	}
+}
+
+# various DoS attacks against the chunk parser:
+{
+	local $SIG{PIPE} = 'IGNORE';
+	my $conn = conn_for($sock, '1.1 chunk header excessive');
+	$conn->write("PUT /sha1 HTTP/1.1\r\nTransfer-Encoding:chunked\r\n\r\n");
+	my $n = 0;
+	my $w;
+	while ($w = $conn->write('ffffffff')) {
+		$n += $w;
+	}
+	ok($!, 'got error set in $!');
+	is($w, undef, 'write error happened');
+	ok($n > 0, 'was able to write');
+	my $r = $conn->read(my $buf, 66666);
+	ok($r > 0, 'got non-empty response');
+	like($buf, qr!HTTP/1\.\d 400 !, 'got 400 response');
+
+	$conn = conn_for($sock, '1.1 chunk trailer excessive');
+	$conn->write("PUT /sha1 HTTP/1.1\r\nTransfer-Encoding:chunked\r\n\r\n");
+	is($conn->syswrite("1\r\na"), 4, 'wrote first header + chunk');
+	delay();
+	$n = 0;
+	while ($w = $conn->write("\r")) {
+		$n += $w;
+	}
+	ok($!, 'got error set in $!');
+	ok($n > 0, 'wrote part of chunk end (\r)');
+	$r = $conn->read($buf, 66666);
+	ok($r > 0, 'got non-empty response');
+	like($buf, qr!HTTP/1\.\d 400 !, 'got 400 response');
+}
+
+{
+	my $conn = conn_for($sock, '1.1 chunked close trickle');
+	$conn->write("PUT /sha1 HTTP/1.1\r\nConnection:close\r\n");
+	$conn->write("Transfer-encoding: chunked\r\n\r\n");
+	foreach my $x ('a'..'z') {
+		delay();
+		$conn->write('1');
+		delay();
+		$conn->write("\r");
+		delay();
+		$conn->write("\n");
+		delay();
+		$conn->write($x);
+		delay();
+		$conn->write("\r");
+		delay();
+		$conn->write("\n");
+	}
+	$conn->write('0');
+	delay();
+	$conn->write("\r");
+	delay();
+	$conn->write("\n");
+	delay();
+	$conn->write("\r");
+	delay();
+	$conn->write("\n");
+	delay();
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, '1.1 chunked close');
+	$conn->write("PUT /sha1 HTTP/1.1\r\nConnection:close\r\n");
+	my $xlen = sprintf('%x', $len);
+	$conn->write("Transfer-Encoding: chunked\r\n\r\n$xlen\r\n" .
+		"$str\r\n0\r\n\r\n");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'chunked body + pipeline');
+	$conn->write("PUT /sha1 HTTP/1.1\r\n" .
+			"Transfer-Encoding: chunked\r\n");
+	delay();
+	$conn->write("\r\n1\r\n");
+	delay();
+	$conn->write('a');
+	delay();
+	$conn->write("\r\n0\r\n\r\nPUT /sha1 HTTP/1.1\r\n");
+	delay();
+
+	my $buf = '';
+	until ($buf =~ /\r\n\r\n[a-f0-9]{40}\z/) {
+		$conn->sysread(my $tmp, 4096);
+		$buf .= $tmp;
+	}
+	my ($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr/\r\nContent-Length: 40\r\n/s, 'got expected length');
+	is($body, sha1_hex('a'), 'read expected body');
+
+	$conn->write("Connection: close\r\n");
+	$conn->write("Content-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'trickle header, one-shot body + pipeline');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n" .
+			"Connection: keep-alive\r\n");
+	delay();
+	$conn->write("Content-Length: $len\r\n\r\n${str}PUT");
+	my $buf = '';
+	until ($buf =~ /\r\n\r\n[a-f0-9]{40}\z/) {
+		$conn->sysread(my $tmp, 4096);
+		$buf .= $tmp;
+	}
+	my ($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr/\r\nContent-Length: 40\r\n/s, 'got expected length');
+	is($body, sha1_hex($str), 'read expected body');
+
+	$conn->write(" /sha1 HTTP/1.0\r\nContent-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'trickle body');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n");
+	$conn->write("Content-Length: $len\r\n\r\n");
+	my $beg = substr($str, 0, 10);
+	my $end = substr($str, 10);
+	is($beg . $end, $str, 'substr setup correct');
+	delay();
+	$conn->write($beg);
+	delay();
+	$conn->write($end);
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'one-shot write');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n" .
+			"Content-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, 'trickle header, one-shot body');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n");
+	delay();
+	$conn->write("Content-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, '1.1 Connnection: close');
+	$conn->write("PUT /sha1 HTTP/1.1\r\nConnection:close\r\n");
+	delay();
+	$conn->write("Content-Length: $len\r\n\r\n$str");
+	$check_self->($conn);
+}
+
+{
+	my $conn = conn_for($sock, '1.1 pipeline start');
+	$conn->write("PUT /sha1 HTTP/1.1\r\n\r\nPUT");
+	my $buf = '';
+	until ($buf =~ /\r\n\r\n[a-f0-9]{40}\z/) {
+		$conn->sysread(my $tmp, 4096);
+		$buf .= $tmp;
+	}
+	my ($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr/\r\nContent-Length: 40\r\n/s, 'got expected length');
+	is($body, sha1_hex(''), 'read expected body');
+
+	# 2nd request
+	$conn->write(" /sha1 HTTP/1.1\r\n\r\n");
+	$buf = '';
+	until ($buf =~ /\r\n\r\n[a-f0-9]{40}\z/) {
+		$conn->sysread(my $tmp, 4096);
+		$buf .= $tmp;
+	}
+	($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr/\r\nContent-Length: 40\r\n/s, 'got expected length');
+	is($body, sha1_hex(''), 'read expected body #2');
+}
+
+done_testing();
+
+1;
diff --git a/t/httpd.t b/t/httpd.t
new file mode 100644
index 0000000..ad636fc
--- /dev/null
+++ b/t/httpd.t
@@ -0,0 +1,119 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+use strict;
+use warnings;
+use Test::More;
+
+foreach my $mod (qw(Plack::Util Plack::Request Plack::Builder Danga::Socket
+			HTTP::Parser::XS HTTP::Date HTTP::Status)) {
+	eval "require $mod";
+	plan skip_all => "$mod missing for httpd.t" if $@;
+}
+use File::Temp qw/tempdir/;
+use Cwd qw/getcwd/;
+use IO::Socket;
+use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
+use Socket qw(SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
+use IPC::Run;
+
+# FIXME: too much setup
+my $tmpdir = tempdir(CLEANUP => 1);
+my $home = "$tmpdir/pi-home";
+my $err = "$tmpdir/stderr.log";
+my $out = "$tmpdir/stdout.log";
+my $pi_home = "$home/.public-inbox";
+my $pi_config = "$pi_home/config";
+my $maindir = "$tmpdir/main.git";
+my $main_bin = getcwd()."/t/main-bin";
+my $main_path = "$main_bin:$ENV{PATH}"; # for spamc ham mock
+my $group = 'test-httpd';
+my $addr = $group . '@example.com';
+my $cfgpfx = "publicinbox.$group";
+my $failbox = "$home/fail.mbox";
+local $ENV{PI_EMERGENCY} = $failbox;
+my $mda = 'blib/script/public-inbox-mda';
+my $httpd = 'blib/script/public-inbox-httpd';
+my $init = 'blib/script/public-inbox-init';
+
+my %opts = (
+	LocalAddr => '127.0.0.1',
+	ReuseAddr => 1,
+	Proto => 'tcp',
+	Type => SOCK_STREAM,
+	Listen => 1024,
+);
+my $sock = IO::Socket::INET->new(%opts);
+my $pid;
+END { kill 'TERM', $pid if defined $pid };
+{
+	local $ENV{HOME} = $home;
+	ok(!system($init, $group, $maindir, 'http://example.com/', $addr),
+		'init ran properly');
+
+	# ensure successful message delivery
+	{
+		local $ENV{ORIGINAL_RECIPIENT} = $addr;
+		my $in = <<EOF;
+From: Me <me\@example.com>
+To: You <you\@example.com>
+Cc: $addr
+Message-Id: <nntp\@example.com>
+Subject: hihi
+Date: Thu, 01 Jan 1970 06:06:06 +0000
+
+nntp
+EOF
+		local $ENV{PATH} = $main_path;
+		IPC::Run::run([$mda], \$in);
+		is(0, $?, 'ran MDA correctly');
+	}
+	ok($sock, 'sock created');
+	$! = 0;
+	my $fl = fcntl($sock, F_GETFD, 0);
+	ok(! $!, 'no error from fcntl(F_GETFD)');
+	is($fl, FD_CLOEXEC, 'cloexec set by default (Perl behavior)');
+	$pid = fork;
+	if ($pid == 0) {
+		use POSIX qw(dup2);
+		# pretend to be systemd
+		fcntl($sock, F_SETFD, $fl &= ~FD_CLOEXEC);
+		dup2(fileno($sock), 3) or die "dup2 failed: $!\n";
+		$ENV{LISTEN_PID} = $$;
+		$ENV{LISTEN_FDS} = 1;
+		exec $httpd, "--stdout=$out", "--stderr=$err";
+		die "FAIL: $!\n";
+	}
+	ok(defined $pid, 'forked httpd process successfully');
+	$! = 0;
+	fcntl($sock, F_SETFD, $fl |= FD_CLOEXEC);
+	ok(! $!, 'no error from fcntl(F_SETFD)');
+	my $host = $sock->sockhost;
+	my $port = $sock->sockport;
+	my $conn = IO::Socket::INET->new(PeerAddr => $host,
+				PeerPort => $port,
+				Proto => 'tcp',
+				Type => SOCK_STREAM);
+	ok($conn, 'connected');
+	ok($conn->write("GET / HTTP/1.0\r\n\r\n"), 'wrote data to socket');
+	{
+		my $buf;
+		ok($conn->read($buf, 4096), 'read some bytes');
+		like($buf, qr!\AHTTP/1\.[01] 404\b!, 'got 404 response');
+		is($conn->read($buf, 1), 0, "EOF");
+	}
+
+	is(system(qw(git clone -q --mirror),
+			"http://$host:$port/$group", "$tmpdir/clone.git"),
+		0, 'clone successful');
+	ok(kill('TERM', $pid), 'killed httpd');
+	$pid = undef;
+	waitpid(-1, 0);
+
+	is(system('git', "--git-dir=$tmpdir/clone.git",
+		  qw(fsck --no-verbose)), 0,
+		'fsck on cloned directory successful');
+}
+
+done_testing();
+
+1;
-- 
EW