qpsmtpd-async

#!/usr/bin/perl

use lib "./lib";

BEGIN {
    delete $ENV{ENV};
    delete $ENV{BASH_ENV};
    $ENV{PATH} = '/bin:/usr/bin:/var/qmail/bin:/usr/local/bin';
}

# Profiling - requires Devel::Profiler 0.05
#BEGIN { $Devel::Profiler::NO_INIT = 1; }
#use Devel::Profiler;

use strict;
use vars qw($DEBUG);
use FindBin qw();

# TODO: need to make this taint friendly
use lib "$FindBin::Bin/lib";
use Danga::Socket;
use Danga::Client;
use Qpsmtpd::PollServer;
use Qpsmtpd::ConfigServer;
use Qpsmtpd::Constants;
use IO::Socket;
use Carp;
use POSIX qw(WNOHANG);
use Getopt::Long;
use List::Util qw(shuffle);

$|++;

use Socket
  qw(SOMAXCONN IPPROTO_TCP SO_KEEPALIVE TCP_NODELAY SOL_SOCKET AF_UNIX SOCK_STREAM PF_UNSPEC);

$SIG{'PIPE'} = "IGNORE";    # handled manually

$DEBUG = 0;

my $CONFIG_PORT      = 20025;
my $CONFIG_LOCALADDR = '127.0.0.1';

my $PORT      = 2525;
my $LOCALADDR = '0.0.0.0';
my $PROCS     = 1;
my $USER      = (getpwuid $>)[0];    # user to suid to
$USER = "smtpd" if $USER eq "root";
my $PAUSED    = 0;
my $NUMACCEPT = 20;
my $PID_FILE  = '';
my $ACCEPT_RSET;
my $DETACH;                          # daemonize on startup

# make sure we don't spend forever doing accept()
use constant ACCEPT_MAX => 1000;

sub reset_num_accept {
    $NUMACCEPT = 20;
}

sub help {
    print <<EOT;
Usage:
    qpsmtpd [OPTIONS]

Options:
 -l, --listen-address addr : listen on a specific address; default 0.0.0.0
 -p, --port P              : listen on a specific port; default 2525
 --config-port             : config server port; default 20025
 -u, --user U              : run as a particular user; defualt 'smtpd'
 -j, --procs J             : spawn J processes; default 1
 -d, --detach              : detach from controlling terminal (daemonize)
     --pid-file P          : print main servers PID to file P
 
 -h, --help                : this page
     --use-poll            : force use of poll() instead of epoll()/kqueue()
EOT
    exit(0);
}

GetOptions(
           'p|port=i'           => \$PORT,
           'l|listen-address=s' => \$LOCALADDR,
           'j|procs=i'          => \$PROCS,
           'v|verbose+'         => \$DEBUG,
           'u|user=s'           => \$USER,
           'pid-file=s'         => \$PID_FILE,
           'd|detach'           => \$DETACH,
           'h|help'             => \&help,
           'config-port=i'      => \$CONFIG_PORT,
          )
  || help();

# detaint the commandline
if ($PORT =~ /^(\d+)$/) { $PORT = $1 }
else                    { &help }
if ($LOCALADDR =~ /^([\d\w\-.]+)$/) { $LOCALADDR = $1 }
else                                { &help }
if ($USER =~ /^([\w\-]+)$/) { $USER = $1 }
else                        { &help }
if ($PROCS =~ /^(\d+)$/) { $PROCS = $1 }
else                     { &help }

sub force_poll {
    $Danga::Socket::HaveEpoll  = 0;
    $Danga::Socket::HaveKQueue = 0;
}

my $POLL = "with "
  . (
       $Danga::Socket::HaveEpoll  ? "epoll()"
     : $Danga::Socket::HaveKQueue ? "kqueue()"
     :                              "poll()"
    );

my $SERVER;
my $CONFIG_SERVER;

use constant READY      => 1;
use constant ACCEPTING  => 2;
use constant RESTARTING => 999;

my %childstatus = ();

if ($PID_FILE && -r $PID_FILE) {
    open PID, "<$PID_FILE"
      or die "open_pidfile $PID_FILE: $!\n";
    my $running_pid = <PID> || '';
    chomp $running_pid;
    if ($running_pid =~ /^(\d+)/) {
        if (kill 0, $running_pid) {
            die "Found an already running qpsmtpd with pid $running_pid.\n";
        }
    }
    close(PID);
}

run_as_server();
exit(0);

sub _fork {
    my $pid = fork;
    if (!defined($pid)) { die "Cannot fork: $!" }
    return $pid if $pid;

    # Fixup Net::DNS randomness after fork
    srand($$ ^ time);

    local $^W;
    delete $INC{'Net/DNS/Header.pm'};
    require Net::DNS::Header;

    # cope with different versions of Net::DNS
    eval {
        $Net::DNS::Resolver::global{id} = 1;
        $Net::DNS::Resolver::global{id} =
          int(rand(Net::DNS::Resolver::MAX_ID()));

        # print "Next DNS ID: $Net::DNS::Resolver::global{id}\n";
    };
    if ($@) {

        # print "Next DNS ID: " . Net::DNS::Header::nextid() . "\n";
    }

    # Fixup lost kqueue after fork
    $Danga::Socket::HaveKQueue = undef;
}

sub spawn_child {
    my $plugin_loader = shift || Qpsmtpd::SMTP->new;

    socketpair(my $reader, my $writer, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
      || die "Unable to create a pipe";
    $writer->autoflush(1);
    $reader->autoflush(1);

    if (my $pid = _fork) {
        $childstatus{$pid} = $writer;
        return $pid;
    }

    $SIG{CHLD} = $SIG{INT} = $SIG{TERM} = 'DEFAULT';
    $SIG{PIPE} = 'IGNORE';
    $SIG{HUP}  = 'IGNORE';

    close $CONFIG_SERVER;

    Qpsmtpd::PollServer->Reset;

    Qpsmtpd::PollServer->OtherFds(
                            fileno($reader) => sub { command_handler($reader) },
                            fileno($SERVER) => \&accept_handler,);

    $ACCEPT_RSET = Danga::Socket->AddTimer(30, \&reset_num_accept);

    $plugin_loader->run_hooks('post-fork');

    Qpsmtpd::PollServer->EventLoop();
    exit;
}

# Note this is broken on KQueue because it requires that it handle signals itself or it breaks the event loop.
sub sig_hup {
    for my $writer (values %childstatus) {
        print $writer "hup\n";
    }
}

sub sig_chld {
    my $spawn_count = 0;
    while ((my $child = waitpid(-1, WNOHANG)) > 0) {
        if (!defined $childstatus{$child}) {
            next;
        }

        last unless $child > 0;
        print "SIGCHLD: child $child died\n";
        delete $childstatus{$child};
        $spawn_count++;
    }
    if ($spawn_count) {
        for (1 .. $spawn_count) {

            # restart a new child if in poll server mode
            my $pid = spawn_child();
        }
    }
    $SIG{CHLD} = \&sig_chld;
}

sub HUNTSMAN {
    $SIG{CHLD} = 'DEFAULT';
    kill 'INT' => keys %childstatus;
    if ($PID_FILE && -e $PID_FILE) {
        unlink $PID_FILE or ::log(LOGERROR, "unlink: $PID_FILE: $!");
    }
    exit(0);
}

sub run_as_server {

    # establish SERVER socket, bind and listen.
    $SERVER = IO::Socket::INET->new(
                                    LocalPort => $PORT,
                                    LocalAddr => $LOCALADDR,
                                    Type      => SOCK_STREAM,
                                    Proto     => IPPROTO_TCP,
                                    Blocking  => 0,
                                    Reuse     => 1,
                                    Listen    => SOMAXCONN
                                   )
      or die "Error creating server $LOCALADDR:$PORT : $@\n";

    IO::Handle::blocking($SERVER, 0);
    binmode($SERVER, ':raw');

    $CONFIG_SERVER =
      IO::Socket::INET->new(
                            LocalPort => $CONFIG_PORT,
                            LocalAddr => $CONFIG_LOCALADDR,
                            Type      => SOCK_STREAM,
                            Proto     => IPPROTO_TCP,
                            Blocking  => 0,
                            Reuse     => 1,
                            Listen    => 1
                           )
      or die "Error creating server $CONFIG_LOCALADDR:$CONFIG_PORT : $@\n";

    IO::Handle::blocking($CONFIG_SERVER, 0);
    binmode($CONFIG_SERVER, ':raw');

    # Drop priviledges
    my (undef, undef, $quid, $qgid) = getpwnam $USER
      or die "unable to determine uid/gid for $USER\n";
    my $groups = "$qgid $qgid";
    while (my (undef, undef, $gid, $members) = getgrent) {
        my @m = split(/ /, $members);
        if (grep { $_ eq $USER } @m) {
            $groups .= " $gid";
        }
    }
    endgrent;
    $) = $groups;
    POSIX::setgid($qgid)
      or die "unable to change gid: $!\n";
    POSIX::setuid($quid)
      or die "unable to change uid: $!\n";
    $> = $quid;

    # Load plugins here
    my $plugin_loader = Qpsmtpd::SMTP->new();
    $plugin_loader->load_plugins;

    if ($DETACH) {
        open STDIN,  '/dev/null'  or die "/dev/null: $!";
        open STDOUT, '>/dev/null' or die "/dev/null: $!";
        open STDERR, '>&STDOUT'   or die "open(stderr): $!";
        defined(my $pid = fork) or die "fork: $!";
        exit 0 if $pid;
        POSIX::setsid or die "setsid: $!";
    }

    if ($PID_FILE) {
        open PID, ">$PID_FILE" || die "$PID_FILE: $!";
        print PID $$, "\n";
        close PID;
    }

    $plugin_loader->log(LOGINFO,
                            'Running as user '
                          . (getpwuid($>) || $>)
                          . ', group '
                          . (getgrgid($)) || $))
                       );

    $SIG{INT} = $SIG{TERM} = \&HUNTSMAN;

######################
    # more Profiling code

=pod
    $plugin_loader->run_hooks('post-fork');

    Devel::Profiler->set_options(
        bad_subs => [qw(Danga::Socket::EventLoop)],
        sub_filter => sub {
                        my ($pkg, $sub) = @_;
                        return 0 if $sub eq 'AUTOLOAD';
                        return 0 if $pkg =~ /ParaDNS::XS/;
                        return 1; 
                      },
    );
    Devel::Profiler->init();
    
    Qpsmtpd::PollServer->OtherFds(
        fileno($SERVER) => \&accept_handler,
        fileno($CONFIG_SERVER) => \&config_handler,    ); 

    Qpsmtpd::PollServer->EventLoop;
    exit;
=cut

#####################

    for (1 .. $PROCS) {
        my $pid = spawn_child($plugin_loader);
    }
    $plugin_loader->log(LOGDEBUG,
                        "Listening on $PORT with $PROCS children $POLL");
    $SIG{CHLD} = \&sig_chld;
    $SIG{HUP}  = \&sig_hup;

    Qpsmtpd::PollServer->OtherFds(fileno($CONFIG_SERVER) => \&config_handler,);

    Qpsmtpd::PollServer->EventLoop;

    exit;

}

sub config_handler {
    my $csock = $CONFIG_SERVER->accept();
    if (!$csock) {

        # warn("accept failed on config server: $!");
        return;
    }
    binmode($csock, ':raw');

    printf("Config server connection\n") if $DEBUG;

    IO::Handle::blocking($csock, 0);
    setsockopt($csock, IPPROTO_TCP, TCP_NODELAY, pack("l", 1)) or die;

    my $client = Qpsmtpd::ConfigServer->new($csock);
    $client->watch_read(1);
    return;
}

sub command_handler {
    my $reader = shift;

    chomp(my $command = <$reader>);

    #print "Got command: $command\n";

    my $real_command = "cmd_$command";

    no strict 'refs';
    $real_command->();
}

sub cmd_hup {

    # clear cache
    print "Clearing cache\n";
    Qpsmtpd::clear_config_cache();

    # should also reload modules... but can't do that yet.
}

# Accept all new connections
sub accept_handler {
    for (1 .. $NUMACCEPT) {
        return unless _accept_handler();
    }

    # got here because we have accept's left.
    # So double the number we accept next time.
    $NUMACCEPT *= 2;
    $NUMACCEPT = ACCEPT_MAX if $NUMACCEPT > ACCEPT_MAX;
    $ACCEPT_RSET->cancel if defined $ACCEPT_RSET;
    $ACCEPT_RSET = Danga::Socket->AddTimer(30, \&reset_num_accept);
}

use Errno qw(EAGAIN EWOULDBLOCK);

sub _accept_handler {
    my $csock = $SERVER->accept();
    if (!$csock) {

        # warn("accept() failed: $!");
        return;
    }
    binmode($csock, ':raw');

    printf("Listen child making a Qpsmtpd::PollServer for %d.\n",
           fileno($csock))
      if $DEBUG;

    IO::Handle::blocking($csock, 0);

    #setsockopt($csock, IPPROTO_TCP, TCP_NODELAY, pack("l", 1)) or die;

    #print "Got connection\n";
    my $client = Qpsmtpd::PollServer->new($csock);

    if ($PAUSED) {
        $client->write("451 Sorry, this server is currently paused\r\n");
        $client->close;
        return 1;
    }

    $client->process_line("Connect\n");
    $client->watch_read(1);
    $client->pause_read();
    return 1;
}

########################################################################

sub log {
    my ($level, $message) = @_;

    # $level not used yet.  this is reimplemented from elsewhere anyway
    warn("$$ fd:? $message\n");
}

sub pause {
    my ($pause) = @_;
    $PAUSED = $pause;
}