[svn:qpsmtpd] rev 496 - in trunk: . lib lib/Qpsmtpd plugins plugins/queue

[email protected] 11 Jul 2005 19:10:49 -0000
Newsgroups perl.cvs.qpsmtpd
Message-ID <[email protected]>
Author: msergeant
Date: Mon Jul 11 12:10:49 2005
New Revision: 496

Added:
   trunk/lib/Qpsmtpd/ConfigServer.pm
      - copied unchanged from rev 485, branches/high_perf/lib/Qpsmtpd/ConfigServer.pm
   trunk/lib/Qpsmtpd/PollServer.pm
      - copied, changed from rev 485, branches/high_perf/lib/Qpsmtpd/PollServer.pm
   trunk/plugins/stats
      - copied unchanged from rev 485, branches/high_perf/plugins/stats
Removed:
   trunk/lib/Qpsmtpd/SelectServer.pm
   trunk/qpsmtpd-server
Modified:
   trunk/   (props changed)
   trunk/lib/Qpsmtpd.pm
   trunk/lib/Qpsmtpd/Constants.pm
   trunk/lib/Qpsmtpd/Plugin.pm
   trunk/lib/Qpsmtpd/SMTP.pm
   trunk/plugins/check_earlytalker
   trunk/plugins/dnsbl
   trunk/plugins/queue/qmail-queue
   trunk/plugins/require_resolvable_fromhost
   trunk/plugins/rhsbl
   trunk/plugins/tls
   trunk/qpsmtpd
   trunk/qpsmtpd-forkserver
Log:
MERGE r386:r480 FROM https://svn.perl.org/qpsmtpd/branches/high_perf
High perf branch merge and fixes


Modified: trunk/lib/Qpsmtpd.pm
==============================================================================
--- trunk/lib/Qpsmtpd.pm	(original)
+++ trunk/lib/Qpsmtpd.pm	Mon Jul 11 12:10:49 2005
@@ -167,9 +167,15 @@ sub _config_from_file {
   return wantarray ? @config : $config[0];
 }
 
+our $HOOKS;
+
 sub load_plugins {
   my $self = shift;
-  
+
+  if ($HOOKS) {
+      return $self->{hooks} = $HOOKS;
+  }
+
   $self->log(LOGWARN, "Plugins already loaded") if $self->{hooks};
   $self->{hooks} = {};
   
@@ -180,6 +186,8 @@ sub load_plugins {
 
   @plugins = $self->_load_plugins($dir, @plugins);
   
+  $HOOKS = $self->{hooks};
+  
   return @plugins;
 }
 
@@ -252,51 +260,21 @@ sub transaction {
 
 sub run_hooks {
   my ($self, $hook) = (shift, shift);
+  if ($self->{_continuation} && $hook ne "logging") {
+    die "Continuations in progress from previous hook (this is the $hook hook)";
+  }
   my $hooks = $self->{hooks};
   if ($hooks->{$hook}) {
     my @r;
-    for my $code (@{$hooks->{$hook}}) {
-      if ( $hook eq 'logging' ) { # without calling $self->log()
-        eval { (@r) = $code->{code}->($self, $self->transaction, @_); };
-        $@ and warn("FATAL LOGGING PLUGIN ERROR: ", $@) and next;
+    my @local_hooks = @{$hooks->{$hook}};
+    while (@local_hooks) {
+      my $code = shift @local_hooks;
+      @r = $self->run_hook($hook, $code, @_);
+      next unless @r;
+      if ($r[0] == CONTINUATION) {
+        $self->disable_read() if $self->isa('Danga::Client');
+        $self->{_continuation} = [$hook, [@_], @local_hooks];
       }
-      else {
-        $self->varlog(LOGINFO, $hook, $code->{name});
-        eval { (@r) = $code->{code}->($self, $self->transaction, @_); };
-        $@ and $self->log(LOGCRIT, "FATAL PLUGIN ERROR: ", $@) and next;
-
-        !defined $r[0]
-          and $self->log(LOGERROR, "plugin ".$code->{name}
-                         ." running the $hook hook returned undef!")
-          and next;
-
-        if ($self->transaction) {
-          my $tnotes = $self->transaction->notes( $code->{name} );
-          $tnotes->{"hook_$hook"}->{'return'} = $r[0]
-            if (!defined $tnotes || ref $tnotes eq "HASH");
-        } else {
-          my $cnotes = $self->connection->notes( $code->{name} );
-          $cnotes->{"hook_$hook"}->{'return'} = $r[0]
-            if (!defined $cnotes || ref $cnotes eq "HASH");
-        }
-
-        # should we have a hook for "OK" too?
-        if ($r[0] == DENY or $r[0] == DENYSOFT or
-            $r[0] == DENY_DISCONNECT or $r[0] == DENYSOFT_DISCONNECT)
-        {
-          $r[1] = "" if not defined $r[1];
-          $self->log(LOGDEBUG, "Plugin ".$code->{name}.
-	    ", hook $hook returned ".return_code($r[0]).", $r[1]");
-          $self->run_hooks("deny", $code->{name}, $r[0], $r[1]) unless ($hook eq "deny");
-        } else {
-          $r[1] = "" if not defined $r[1];
-          $self->log(LOGDEBUG, "Plugin ".$code->{name}.
-	    ", hook $hook returned ".return_code($r[0]).", $r[1]");
-          $self->run_hooks("ok", $code->{name}, $r[0], $r[1]) unless ($hook eq "ok");
-	}
-
-      }
-
       last unless $r[0] == DECLINED;
     }
     $r[0] = DECLINED if not defined $r[0];
@@ -305,6 +283,77 @@ sub run_hooks {
   return (0, '');
 }
 
+sub finish_continuation {
+  my ($self) = @_;
+  die "No continuation in progress" unless $self->{_continuation};
+  $self->enable_read() if $self->isa('Danga::Client');
+  my $todo = $self->{_continuation};
+  $self->{_continuation} = undef;
+  my $hook = shift @$todo || die "No hook in the continuation";
+  my $args = shift @$todo || die "No hook args in the continuation";
+  my @r;
+  while (@$todo) {
+    my $code = shift @$todo;
+    @r = $self->run_hook($hook, $code, @$args);
+    if ($r[0] == CONTINUATION) {
+      $self->disable_read() if $self->isa('Danga::Client');
+      $self->{_continuation} = [$hook, $args, @$todo];
+      return @r;
+    }
+    last unless $r[0] == DECLINED;
+  }
+  $r[0] = DECLINED if not defined $r[0];
+  my $responder = $hook . "_respond";
+  if (my $meth = $self->can($responder)) {
+    warn("continuation finished on $self\n");
+    return $meth->($self, $r[0], $r[1], @$args);
+  }
+  die "No ${hook}_respond method";
+}
+
+sub run_hook {
+  my ($self, $hook, $code, @args) = @_;
+  my @r;
+  if ( $hook eq 'logging' ) { # without calling $self->log()
+    eval { (@r) = $code->{code}->($self, $self->transaction, @_); };
+    $@ and warn("FATAL LOGGING PLUGIN ERROR: ", $@) and next;
+  }
+  else {
+    $self->varlog(LOGINFO, $hook, $code->{name});
+    print STDERR "plugin $hook $code->{name} 1\n";
+    eval { (@r) = $code->{code}->($self, $self->transaction, @args); };
+    print STDERR "plugin $hook $code->{name} 2\n";
+    
+    $@ and $self->log(LOGCRIT, "FATAL PLUGIN ERROR: ", $@) and return;
+  
+    !defined $r[0]
+      and $self->log(LOGERROR, "plugin ".$code->{name}
+                     ."running the $hook hook returned undef!")
+        and return;
+  
+    if ($self->transaction) {
+      my $tnotes = $self->transaction->notes( $code->{name} );
+      $tnotes->{"hook_$hook"}->{'return'} = $r[0]
+        if (!defined $tnotes || ref $tnotes eq "HASH");
+    } else {
+      my $cnotes = $self->connection->notes( $code->{name} );
+      $cnotes->{"hook_$hook"}->{'return'} = $r[0]
+        if (!defined $cnotes || ref $cnotes eq "HASH");
+    }
+  
+    # should we have a hook for "OK" too?
+    if ($r[0] == DENY or $r[0] == DENYSOFT or
+        $r[0] == DENY_DISCONNECT or $r[0] == DENYSOFT_DISCONNECT)
+    {
+      $r[1] = "" if not defined $r[1];
+      $self->log(LOGDEBUG, "Plugin $code->{name}, hook $hook returned $r[0], $r[1]");
+      $self->run_hooks("deny", $code->{name}, $r[0], $r[1]) unless ($hook eq "deny");
+    }
+  
+  }
+  return @r;
+}
+
 sub _register_hook {
   my $self = shift;
   my ($hook, $code, $unshift) = @_;

Modified: trunk/lib/Qpsmtpd/Constants.pm
==============================================================================
--- trunk/lib/Qpsmtpd/Constants.pm	(original)
+++ trunk/lib/Qpsmtpd/Constants.pm	Mon Jul 11 12:10:49 2005
@@ -4,27 +4,28 @@ require Exporter;
 
 # log levels
 my %log_levels = (
-	LOGDEBUG   => 7,
-	LOGINFO    => 6,
-	LOGNOTICE  => 5,
-	LOGWARN    => 4,
-	LOGERROR   => 3,
-	LOGCRIT    => 2,
-	LOGALERT   => 1,
-	LOGEMERG   => 0,
-	LOGRADAR   => 0,
+        LOGDEBUG   => 7,
+        LOGINFO    => 6,
+        LOGNOTICE  => 5,
+        LOGWARN    => 4,
+        LOGERROR   => 3,
+        LOGCRIT    => 2,
+        LOGALERT   => 1,
+        LOGEMERG   => 0,
+        LOGRADAR   => 0,
 );
 
 # return codes
 my %return_codes = (
-	OK       => 900,
-	DENY     => 901,   # 550
-	DENYSOFT => 902,   # 450
-	DENYHARD => 903,   # 550 + disconnect  (deprecated in 0.29)
-	DENY_DISCONNECT     => 903, # 550 + disconnect
-	DENYSOFT_DISCONNECT => 904, # 450 + disconnect
-	DECLINED => 909,
-	DONE     => 910,
+        OK                     => 900,
+        DENY                   => 901,   # 550
+        DENYSOFT               => 902,   # 450
+        DENYHARD               => 903,   # 550 + disconnect  (deprecated in 0.29)
+        DENY_DISCONNECT        => 903,   # 550 + disconnect
+        DENYSOFT_DISCONNECT    => 904,   # 450 + disconnect
+        DECLINED               => 909,
+        DONE                   => 910,
+        CONTINUATION           => 911,
 );
 
 use vars qw(@ISA @EXPORT);
@@ -42,24 +43,24 @@ foreach (keys %log_levels ) {
 sub return_code {
     my $test = shift;
     if ( $test =~ /^\d+$/ ) { # need to return the textural form
-	foreach ( keys %return_codes ) {
-	    return $_ if $return_codes{$_} =~ /$test/;
-	}
+        foreach ( keys %return_codes ) {
+            return $_ if $return_codes{$_} =~ /$test/;
+        }
     }
     else { # just return the numeric value
-	return $return_codes{$test};
+        return $return_codes{$test};
     }
 }
 
 sub log_level {
     my $test = shift;
     if ( $test =~ /^\d+$/ ) { # need to return the textural form
-	foreach ( keys %log_levels ) {
-	    return $_ if $log_levels{$_} =~ /$test/;
-	}
+        foreach ( keys %log_levels ) {
+            return $_ if $log_levels{$_} =~ /$test/;
+        }
     }
     else { # just return the numeric value
-	return $log_levels{$test};
+        return $log_levels{$test};
     }
 }
 

Modified: trunk/lib/Qpsmtpd/Plugin.pm
==============================================================================
--- trunk/lib/Qpsmtpd/Plugin.pm	(original)
+++ trunk/lib/Qpsmtpd/Plugin.pm	Mon Jul 11 12:10:49 2005
@@ -37,11 +37,15 @@ sub _register {
   my $self = shift;
   my $qp = shift;
   local $self->{_qp} = $qp;
-  $self->init($qp, @_)     if $self->can('init');
+  $self->init($qp, @_);
   $self->_register_standard_hooks($qp, @_);
-  $self->register($qp, @_) if $self->can('register');
+  $self->register($qp, @_);
 }
 
+# Designed to be overloaded
+sub init {}
+sub register {}
+
 sub qp {
   shift->{_qp};
 }
@@ -61,6 +65,10 @@ sub connection {
   shift->qp->connection;
 }
 
+sub config {
+  shift->qp->config(@_);
+}
+
 sub spool_dir {
   shift->qp->spool_dir;
 }

Copied: trunk/lib/Qpsmtpd/PollServer.pm (from rev 485, branches/high_perf/lib/Qpsmtpd/PollServer.pm)
==============================================================================
--- branches/high_perf/lib/Qpsmtpd/PollServer.pm	(original)
+++ trunk/lib/Qpsmtpd/PollServer.pm	Mon Jul 11 12:10:49 2005
@@ -111,19 +111,14 @@ sub process_line {
         my ($pkg, $file, $line) = caller();
         die "ALARM: ($self->{mode}) $pkg, $file, $line";
     };
-    if( $self->{mode} eq 'connect' ) {
-        eval { $self->_process_line($line) }
-    }
-    else {
-        my $prev = alarm(2); # must process a command in < 2 seconds
-        eval { $self->_process_line($line) };
-        alarm($prev);
-        if ($@) {
-            print STDERR "Error: $@\n";
-            return $self->fault("command failed unexpectedly") if $self->{mode} eq 'cmd';
-            return $self->fault("error processing data lines") if $self->{mode} eq 'data';
-            return $self->fault("unknown error");
-        }
+    my $prev = alarm(2); # must process a command in < 2 seconds
+    eval { $self->_process_line($line) };
+    alarm($prev);
+    if ($@) {
+        print STDERR "Error: $@\n";
+        return $self->fault("command failed unexpectedly") if $self->{mode} eq 'cmd';
+        return $self->fault("error processing data lines") if $self->{mode} eq 'data';
+        return $self->fault("unknown error");
     }
     return;
 }
@@ -168,8 +163,8 @@ sub process_cmd {
     }
     else {
         # No such method - i.e. unrecognized command
-        my ($rc, $msg) = $self->run_hooks("unrecognized_command", $cmd);
-        return $self->unrecognized_command_respond unless $rc == CONTINUATION;
+        my ($rc, $msg) = $self->run_hooks("unrecognized_command", $meth, @params);
+        return $self->unrecognized_command_respond($rc, $msg) unless $rc == CONTINUATION;
         return 1;
     }
 }
@@ -257,8 +252,7 @@ sub data_line {
     if ($line eq ".\r\n") {
         # add received etc.
         $self->{mode} = 'cmd';
-        $self->end_of_data;
-        return;
+        return $self->end_of_data;
     }
 
     # Reject messages that have either bare LF or CR. rjkaes noticed a

Modified: trunk/lib/Qpsmtpd/SMTP.pm
==============================================================================
--- trunk/lib/Qpsmtpd/SMTP.pm	(original)
+++ trunk/lib/Qpsmtpd/SMTP.pm	Mon Jul 11 12:10:49 2005
@@ -51,21 +51,9 @@ sub dispatch {
   $self->{_counter}++; 
 
   if ($cmd !~ /^(\w{1,12})$/ or !exists $self->{_commands}->{$1}) {
-    my ($rc, $msg) = $self->run_hooks("unrecognized_command", $cmd);
-    if ($rc == DENY_DISCONNECT) {
-      $self->respond(521, $msg);
-      $self->disconnect;
-    }
-    elsif ($rc == DENY) {
-      $self->respond(500, $msg);
-    }
-    elsif ($rc == DONE) {
-      1;
-    }
-    else {
-      $self->respond(500, "Unrecognized command");
-    }
-    return 1
+    my ($rc, $msg) = $self->run_hooks("unrecognized_command", $cmd, @_);
+    return $self->unrecognized_command_respond($rc, $msg, @_) unless $rc == CONTINUATION;
+    return 1;
   }
   $cmd = $1;
 
@@ -79,6 +67,20 @@ sub dispatch {
   return;
 }
 
+sub unrecognized_command_respond {
+    my ($self, $rc, $msg) = @_;
+    if ($rc == DENY_DISCONNECT) {
+      $self->respond(521, $msg);
+      $self->disconnect;
+    }
+    elsif ($rc == DENY) {
+      $self->respond(500, $msg);
+    }
+    elsif ($rc != DONE) {
+      $self->respond(500, "Unrecognized command");
+    }
+}
+
 sub fault {
   my $self = shift;
   my ($msg) = shift || "program fault - command not performed";
@@ -92,12 +94,20 @@ sub start_conversation {
     # this should maybe be called something else than "connect", see
     # lib/Qpsmtpd/TcpServer.pm for more confusion.
     my ($rc, $msg) = $self->run_hooks("connect");
+    return $self->connect_respond($rc, $msg) unless $rc == CONTINUATION;
+    return 1;
+}
+
+sub connect_respond {
+    my ($self, $rc, $msg) = @_;
     if ($rc == DENY) {
       $self->respond(550, ($msg || 'Connection from you denied, bye bye.'));
+      $self->disconnect;
       return $rc;
     }
     elsif ($rc == DENYSOFT) {
       $self->respond(450, ($msg || 'Connection from you temporarily denied, bye bye.'));
+      $self->disconnect;
       return $rc;
     }
     elsif ($rc == DONE) {
@@ -124,6 +134,7 @@ sub reset_transaction {
 
 sub connection {
   my $self = shift;
+  @_ and $self->{_connection} = shift;
   return $self->{_connection} || ($self->{_connection} = Qpsmtpd::Connection->new());
 }
 
@@ -136,11 +147,16 @@ sub helo {
   return $self->respond (503, "but you already said HELO ...") if $conn->hello;
 
   my ($rc, $msg) = $self->run_hooks("helo", $hello_host, @stuff);
-  if ($rc == DONE) {
-    # do nothing
-  } elsif ($rc == DENY) {
+  return $self->helo_respond($rc, $msg, $hello_host, @stuff) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub helo_respond {
+  my ($self, $rc, $msg, $hello_host) = @_;
+  if ($rc == DENY) {
     $self->respond(550, $msg);
-  } elsif ($rc == DENYSOFT) {
+  }
+  elsif ($rc == DENYSOFT) {
     $self->respond(450, $msg);
   } elsif ($rc == DENY_DISCONNECT) {
       $self->respond(550, $msg);
@@ -148,11 +164,14 @@ sub helo {
   } elsif ($rc == DENYSOFT_DISCONNECT) {
       $self->respond(450, $msg);
       $self->disconnect;
-  } else {
+  }
+  elsif ($rc != DONE) {
+    my $conn = $self->connection;
     $conn->hello("helo");
     $conn->hello_host($hello_host);
     $self->transaction;
-    $self->respond(250, $self->config('me') ." Hi " . $conn->remote_info . " [" . $conn->remote_ip ."]; I am so happy to meet you.");
+    $self->respond(250, $self->config('me') ." Hi " . $conn->remote_info . 
+                        " [" . $conn->remote_ip ."]; I am so happy to meet you.");
   }
 }
 
@@ -164,11 +183,16 @@ sub ehlo {
   return $self->respond (503, "but you already said HELO ...") if $conn->hello;
 
   my ($rc, $msg) = $self->run_hooks("ehlo", $hello_host, @stuff);
-  if ($rc == DONE) {
-    # do nothing
-  } elsif ($rc == DENY) {
+  return $self->ehlo_respond($rc, $msg, $hello_host, @stuff) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub ehlo_respond {
+  my ($self, $rc, $msg, $hello_host) = @_;
+  if ($rc == DENY) {
     $self->respond(550, $msg);
-  } elsif ($rc == DENYSOFT) {
+  }
+  elsif ($rc == DENYSOFT) {
     $self->respond(450, $msg);
   } elsif ($rc == DENY_DISCONNECT) {
       $self->respond(550, $msg);
@@ -176,7 +200,9 @@ sub ehlo {
   } elsif ($rc == DENYSOFT_DISCONNECT) {
       $self->respond(450, $msg);
       $self->disconnect;
-  } else {
+  }
+  elsif ($rc != DONE) {
+    my $conn = $self->connection;
     $conn->hello("ehlo");
     $conn->hello_host($hello_host);
     $self->transaction;
@@ -241,57 +267,62 @@ sub mail {
   unless ($self->connection->hello) {
     return $self->respond(503, "please say hello first ...");
   }
-  else {
-    my $from_parameter = join " ", @_;
-    $self->log(LOGINFO, "full from_parameter: $from_parameter");
+  
+  my $from_parameter = join " ", @_;
+  $self->log(LOGINFO, "full from_parameter: $from_parameter");
 
-    my ($from) = ($from_parameter =~ m/^from:\s*(<[^>]*>)/i)[0];
+  my ($from) = ($from_parameter =~ m/^from:\s*(<[^>]*>)/i)[0];
 
-    # support addresses without <> ... maybe we shouldn't?
-    ($from) = "<" . ($from_parameter =~ m/^from:\s*(\S+)/i)[0] . ">"
-      unless $from;
-
-    $self->log(LOGALERT, "from email address : [$from]");
-
-    if ($from eq "<>" or $from =~ m/\[undefined\]/ or $from eq "<#@[]>") {
-      $from = Qpsmtpd::Address->new("<>");
-    } 
-    else {
-      $from = (Qpsmtpd::Address->parse($from))[0];
-    }
-    return $self->respond(501, "could not parse your mail from command") unless $from;
+  # support addresses without <> ... maybe we shouldn't?
+  ($from) = "<" . ($from_parameter =~ m/^from:\s*(\S+)/i)[0] . ">"
+    unless $from;
 
-    my ($rc, $msg) = $self->run_hooks("mail", $from);
-    if ($rc == DONE) {
-      return 1;
-    }
-    elsif ($rc == DENY) {
-      $msg ||= $from->format . ', denied';
-      $self->log(LOGINFO, "deny mail from " . $from->format . " ($msg)");
-      $self->respond(550, $msg);
-    }
-    elsif ($rc == DENYSOFT) {
-      $msg ||= $from->format . ', temporarily denied';
-      $self->log(LOGINFO, "denysoft mail from " . $from->format . " ($msg)");
-      $self->respond(450, $msg);
-    }
-    elsif ($rc == DENY_DISCONNECT) {
-      $msg ||= $from->format . ', denied';
-      $self->log(LOGINFO, "deny mail from " . $from->format . " ($msg)");
-      $self->respond(550, $msg);
-      $self->disconnect;
-    }
-    elsif ($rc == DENYSOFT_DISCONNECT) {
-      $msg ||= $from->format . ', temporarily denied';
-      $self->log(LOGINFO, "denysoft mail from " . $from->format . " ($msg)");
-      $self->respond(421, $msg);
-      $self->disconnect;
-    }
-    else { # includes OK
-      $self->log(LOGINFO, "getting mail from ".$from->format);
-      $self->respond(250, $from->format . ", sender OK - how exciting to get mail from you!");
-      $self->transaction->sender($from);
-    }
+  $self->log(LOGALERT, "from email address : [$from]");
+
+  if ($from eq "<>" or $from =~ m/\[undefined\]/ or $from eq "<#@[]>") {
+    $from = Qpsmtpd::Address->new("<>");
+  } 
+  else {
+    $from = (Qpsmtpd::Address->parse($from))[0];
+  }
+  return $self->respond(501, "could not parse your mail from command") unless $from;
+
+  my ($rc, $msg) = $self->run_hooks("mail", $from);
+  return $self->mail_respond($rc, $msg, $from) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub mail_respond {
+  my ($self, $rc, $msg, $from) = @_;
+  if ($rc == DONE) {
+    return 1;
+  }
+  elsif ($rc == DENY) {
+    $msg ||= $from->format . ', denied';
+    $self->log(LOGINFO, "deny mail from " . $from->format . " ($msg)");
+    $self->respond(550, $msg);
+  }
+  elsif ($rc == DENYSOFT) {
+    $msg ||= $from->format . ', temporarily denied';
+    $self->log(LOGINFO, "denysoft mail from " . $from->format . " ($msg)");
+    $self->respond(450, $msg);
+  }
+  elsif ($rc == DENY_DISCONNECT) {
+    $msg ||= $from->format . ', denied';
+    $self->log(LOGINFO, "deny mail from " . $from->format . " ($msg)");
+    $self->respond(550, $msg);
+    $self->disconnect;
+  }
+  elsif ($rc == DENYSOFT_DISCONNECT) {
+    $msg ||= $from->format . ', temporarily denied';
+    $self->log(LOGINFO, "denysoft mail from " . $from->format . " ($msg)");
+    $self->respond(450, $msg);
+    $self->disconnect;
+  }
+  else { # includes OK
+    $self->log(LOGINFO, "getting mail from ".$from->format);
+    $self->respond(250, $from->format . ", sender OK - how exciting to get mail from you!");
+    $self->transaction->sender($from);
   }
 }
 
@@ -308,6 +339,12 @@ sub rcpt {
   return $self->respond(501, "could not parse recipient") unless $rcpt;
 
   my ($rc, $msg) = $self->run_hooks("rcpt", $rcpt);
+  return $self->rcpt_respond($rc, $msg, $rcpt) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub rcpt_respond {
+  my ($self, $rc, $msg, $rcpt) = @_;
   if ($rc == DONE) {
     return 1;
   }
@@ -342,7 +379,6 @@ sub rcpt {
 }
 
 
-
 sub help {
   my $self = shift;
   $self->respond(214, 
@@ -364,6 +400,12 @@ sub vrfy {
   # I also don't think it provides all the proper result codes.
 
   my ($rc, $msg) = $self->run_hooks("vrfy");
+  return $self->vrfy_respond($rc, $msg) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub vrfy_respond {
+  my ($self, $rc, $msg) = @_;
   if ($rc == DONE) {
     return 1;
   }
@@ -391,6 +433,12 @@ sub rset {
 sub quit {
   my $self = shift;
   my ($rc, $msg) = $self->run_hooks("quit");
+  return $self->quit_respond($rc, $msg) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub quit_respond {
+  my ($self, $rc, $msg) = @_;
   if ($rc != DONE) {
     $self->respond(221, $self->config('me') . " closing connection. Have a wonderful day.");
   }
@@ -403,9 +451,17 @@ sub disconnect {
   $self->reset_transaction;
 }
 
+sub disconnect_respond { }
+
 sub data {
   my $self = shift;
   my ($rc, $msg) = $self->run_hooks("data");
+  return $self->data_respond($rc, $msg) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub data_respond {
+  my ($self, $rc, $msg) = @_;
   if ($rc == DONE) {
     return 1;
   }
@@ -523,6 +579,11 @@ sub data {
   $self->respond(552, "Message too big!"),return 1 if $max_size and $size > $max_size;
 
   ($rc, $msg) = $self->run_hooks("data_post");
+  return $self->data_post_respond($rc, $msg) unless $rc == CONTINUATION;
+}
+
+sub data_post_respond {
+  my ($self, $rc, $msg) = @_;
   if ($rc == DONE) {
     return 1;
   }
@@ -538,7 +599,6 @@ sub data {
 
   # DATA is always the end of a "transaction"
   return $self->reset_transaction;
-
 }
 
 sub getline {
@@ -554,6 +614,12 @@ sub queue {
   my ($self, $transaction) = @_;
 
   my ($rc, $msg) = $self->run_hooks("queue");
+  return $self->queue_respond($rc, $msg) unless $rc == CONTINUATION;
+  return 1;
+}
+
+sub queue_respond {
+  my ($self, $rc, $msg) = @_;
   if ($rc == DONE) {
     return 1;
   }
@@ -569,8 +635,6 @@ sub queue {
   else {
     $self->respond(451, $msg || "Queuing declined or disabled; try again later" );
   }
-
-
 }
 
 

Modified: trunk/plugins/check_earlytalker
==============================================================================
--- trunk/plugins/check_earlytalker	(original)
+++ trunk/plugins/check_earlytalker	Mon Jul 11 12:10:49 2005
@@ -44,38 +44,36 @@ and terminating the SMTP connection.
 
 =cut
 
-use IO::Select;
-
-use warnings;
-use strict;
+my $MSG = 'Connecting host started transmitting before SMTP greeting';
 
 sub register {
   my ($self, $qp, @args) = @_;
 
   if (@args % 2) {
-        $self->log(LOGERROR, "Unrecognized/mismatched arguments");
-        return undef;
+	$self->log(LOGERROR, "Unrecognized/mismatched arguments");
+	return undef;
   }
   $self->{_args} = {
-        'wait' => 1,
-        'action' => 'denysoft',
-        'defer-reject' => 0,
-        @args,
+  	'wait' => 1,
+	'action' => 'denysoft',
+	'defer-reject' => 0,
+	@args,
   };
   if ($qp->{conn} && $qp->{conn}->isa('Apache2::Connection')) {
       require APR::Const;
       APR::Const->import(qw(POLLIN SUCCESS));
-      $self->register_hook('connect', 'apr_connect_handler');
+      $self->register_hook('connect', 'hook_connect_apr');
   }
   else {
-      $self->register_hook('connect', 'connect_handler');
+      $self->register_hook('connect', 'hook_connect');
   }
-  $self->register_hook('mail', 'mail_handler')
+  $self->register_hook('connect', 'hook_connect_post');
+  $self->register_hook('mail', 'hook_mail')
     if $self->{_args}->{'defer-reject'};
   1;
 }
 
-sub apr_connect_handler {
+sub hook_connect_apr {
     my ($self, $transaction) = @_;
     
     return DECLINED if ($self->qp->connection->notes('whitelistclient'));
@@ -92,47 +90,55 @@ sub apr_connect_handler {
             $self->qp->connection->notes('earlytalker', 1);
         }
         else {
-            my $msg = 'Connecting host started transmitting before SMTP greeting';
-            return (DENY,$msg) if $self->{_args}->{'action'} eq 'deny';
-            return (DENYSOFT,$msg) if $self->{_args}->{'action'} eq 'denysoft';
+            return (DENY,$MSG) if $self->{_args}->{'action'} eq 'deny';
+            return (DENYSOFT,$MSG) if $self->{_args}->{'action'} eq 'denysoft';
         }
     }
     else {
         $self->log(LOGINFO, "remote host said nothing spontaneous, proceeding");
     }
+    return DECLINED;
 }
 
-sub connect_handler {
+sub hook_connect {
   my ($self, $transaction) = @_;
-  my $in = new IO::Select;
-  my $ip = $self->qp->connection->remote_ip;
-
-  return DECLINED
-      if ($self->qp->connection->notes('whitelistclient'));
+  
+  my $qp = $self->qp;
+  my $conn = $qp->connection;
+  $qp->AddTimer($self->{_args}{'wait'}, sub { read_now($qp, $conn) });
+  return CONTINUATION;
+}
 
-  $in->add(\*STDIN) || return DECLINED;
-  if ($in->can_read($self->{_args}->{'wait'})) {
-    $self->log(LOGNOTICE, "remote host started talking before we said hello [$ip]");
-    if ($self->{_args}->{'defer-reject'}) {
-        $self->qp->connection->notes('earlytalker', 1);
-    } else {
-      my $msg = 'Connecting host started transmitting before SMTP greeting';
-      return (DENY,$msg) if $self->{_args}->{'action'} eq 'deny';
-      return (DENYSOFT,$msg) if $self->{_args}->{'action'} eq 'denysoft';
+sub read_now {
+  my ($qp, $conn) = @_;
+  
+  if (my $data = $qp->read(1024)) {
+    if (length($$data)) {
+      $qp->log(LOGNOTICE, 'remote host started talking before we said hello');
+      $qp->push_back_read($data);
+      $conn->notes('earlytalker', 1);
     }
-  } else {
-    $self->log(LOGINFO, 'remote host said nothing spontaneous, proceeding');
   }
-  return DECLINED;
+  $qp->finish_continuation;
+}
+
+sub hook_connect_post {
+  my ($self, $transaction) = @_;
+  
+  my $conn = $self->qp->connection;
+  return DECLINED unless $conn->notes('earlytalker');
+  return DECLINED if $self->{'defer-reject'};
+  return (DENY,$MSG) if $self->{_args}->{'action'} eq 'deny';
+  return (DENYSOFT,$MSG) if $self->{_args}->{'action'} eq 'denysoft';
+  return DECLINED; # assume action eq 'log'
 }
 
-sub mail_handler {
+sub hook_mail {
   my ($self, $txn) = @_;
-  my $msg = 'Connecting host started transmitting before SMTP greeting';
 
-  return DECLINED unless $self->qp->connection->notes('earlytalker');
-  return (DENY,$msg) if $self->{_args}->{'action'} eq 'deny';
-  return (DENYSOFT,$msg) if $self->{_args}->{'action'} eq 'denysoft';
+  return DECLINED unless $self->connection->notes('earlytalker');
+  return (DENY,$MSG) if $self->{_args}->{'action'} eq 'deny';
+  return (DENYSOFT,$MSG) if $self->{_args}->{'action'} eq 'denysoft';
   return DECLINED;
 }
 

Modified: trunk/plugins/dnsbl
==============================================================================
--- trunk/plugins/dnsbl	(original)
+++ trunk/plugins/dnsbl	Mon Jul 11 12:10:49 2005
@@ -1,20 +1,17 @@
-#!perl -w
+#!/usr/bin/perl -w
 
-sub register {
-  my ($self, $qp, $denial ) = @_;
-  if ( defined $denial and $denial =~ /^disconnect$/i ) {
-    $self->{_dnsbl}->{DENY} = DENY_DISCONNECT;
-  }
-  else {
-    $self->{_dnsbl}->{DENY} = DENY;
-  }
+use Danga::DNS;
 
+sub register {
+  my ($self) = @_;
+  $self->register_hook("connect", "connect_handler");
+  $self->register_hook("connect", "pickup_handler");
 }
 
-sub hook_connect {
+sub connect_handler {
   my ($self, $transaction) = @_;
 
-  my $remote_ip = $self->qp->connection->remote_ip;
+  my $remote_ip = $self->connection->remote_ip;
 
   # perform RBLSMTPD checks to mimic Dan Bernstein's rblsmtpd
   if (defined($ENV{'RBLSMTPD'})) {
@@ -29,163 +26,91 @@ sub hook_connect {
     $self->log(LOGDEBUG, "RBLSMTPD not set for $remote_ip");
   }
 
-  my $allow = grep { s/\.?$/./; $_ eq substr($remote_ip . '.', 0, length $_) } $self->qp->config('dnsbl_allow');
+  my $allow = grep { s/\.?$/./; $_ eq substr($remote_ip . '.', 0, length $_) } $self->config('dnsbl_allow');
   return DECLINED if $allow;
 
-  my %dnsbl_zones = map { (split /:/, $_, 2)[0,1] } $self->qp->config('dnsbl_zones');
+  my %dnsbl_zones = map { (split /:/, $_, 2)[0,1] } $self->config('dnsbl_zones');
   return DECLINED unless %dnsbl_zones;
 
   my $reversed_ip = join(".", reverse(split(/\./, $remote_ip)));
 
-  # we should queue these lookups in the background and just fetch the
-  # results in the first rcpt handler ... oh well.
-
-  my $res = new Net::DNS::Resolver;
-  $res->tcp_timeout(30);
-  $res->udp_timeout(30);
-
-  my $sel = IO::Select->new();
-
+  $self->transaction->notes('pending_dns_queries', scalar(keys(%dnsbl_zones)));
+  my $qp = $self->qp;
   for my $dnsbl (keys %dnsbl_zones) {
     # fix to find A records, if the dnsbl_zones line has a second field 20/1/04 ++msp
     if (defined($dnsbl_zones{$dnsbl})) {
       $self->log(LOGDEBUG, "Checking $reversed_ip.$dnsbl for A record in the background");
-      $sel->add($res->bgsend("$reversed_ip.$dnsbl"));
+      Danga::DNS->new(
+        callback => sub { process_a_result($qp, $dnsbl_zones{$dnsbl}, @_) },
+        host => "$reversed_ip.$dnsbl",
+        type => 'A',
+        client => $self->qp->input_sock,
+      );
     } else {
       $self->log(LOGDEBUG, "Checking $reversed_ip.$dnsbl for TXT record in the background");
-      $sel->add($res->bgsend("$reversed_ip.$dnsbl", "TXT"));
+      Danga::DNS->new(
+        callback => sub { process_txt_result($qp, @_) },
+        host => "$reversed_ip.$dnsbl",
+        type => 'TXT',
+        client => $self->qp->input_sock,
+      );
     }
   }
 
-  $self->qp->connection->notes('dnsbl_sockets', $sel);
-
-  return DECLINED;
+  return CONTINUATION;
 }
 
-sub process_sockets {
-  my ($self) = @_;
-
-  my $conn = $self->qp->connection;
-
-  return $conn->notes('dnsbl') 
-    if $conn->notes('dnsbl');
-
-  my %dnsbl_zones = map { (split /:/, $_, 2)[0,1] } $self->qp->config('dnsbl_zones');
-
-  my $res = new Net::DNS::Resolver;
-  $res->tcp_timeout(30);
-  $res->udp_timeout(30);
-
-  my $sel = $conn->notes('dnsbl_sockets') or return "";
-  my $remote_ip = $self->qp->connection->remote_ip;
-
-  my $result; 
-
-  $self->log(LOGDEBUG, "waiting for dnsbl dns");
-
-  # don't wait more than 8 seconds here
-  my @ready = $sel->can_read(8);
-
-  $self->log(LOGDEBUG, "DONE waiting for dnsbl dns, got " , scalar @ready, " answers ...") ;
-  return '' unless @ready;
-
-  for my $socket (@ready) {
-    my $query = $res->bgread($socket);
-    $sel->remove($socket);
-    undef $socket;
-
-    my $dnsbl;
-
-    if ($query) {
-      my $a_record = 0;
-      foreach my $rr ($query->answer) {
-	$a_record = 1 if $rr->type eq "A";
-	my $name = $rr->name;
-	($dnsbl) = ($name =~ m/(?:\d+\.){4}(.*)/) unless $dnsbl;
-	$dnsbl = $name unless $dnsbl;
-	$self->log(LOGDEBUG, "name ", $rr->name);
-	next unless $rr->type eq "TXT";
-	$self->log(LOGDEBUG, "got txt record");
-	$result = $rr->txtdata and last;
-      }
-      #$a_record and $result = "Blocked by $dnsbl";
-
-      if ($a_record) {
-        if (defined $dnsbl_zones{$dnsbl}) {
-          $result = $dnsbl_zones{$dnsbl};
-          #$result =~ s/%IP%/$ENV{'TCPREMOTEIP'}/g;
-          $result =~ s/%IP%/$remote_ip/g;
-        } else {
-          # shouldn't get here?
-          $result = "Blocked by $dnsbl";
-        }
-      }
-    }
-    else {
-      $self->log(LOGERROR, "$dnsbl query failed: ", $res->errorstring)
-	unless $res->errorstring eq "NXDOMAIN";
-    }
-
-    if ($result) {
-      #kill any other pending I/O
-      $conn->notes('dnsbl_sockets', undef);
-      $result = join("\n", $self->qp->config('dnsbl_rejectmsg'), $result);
-      return $conn->notes('dnsbl', $result);
-    }
-  }
-
-  if ($sel->count) {
-    # loop around if we have dns blacklists left to see results from
-    return $self->process_sockets();
-  }
-
-  # er, the following code doesn't make much sense anymore...
-
-  # if there was more to read; then forget it
-  $conn->notes('dnsbl_sockets', undef);
-
-  return $conn->notes('dnsbl', $result);
+sub process_a_result {
+    my ($qp, $template, $result, $query) = @_;
+    
+    my $pending = $qp->transaction->notes('pending_dns_queries');
+    $qp->transaction->notes('pending_dns_queries', --$pending);
+    
+    warn("Result for A $query: $result\n");
+    if ($result !~ /^\d+\.\d+\.\d+\.\d+$/) {
+        # NXDOMAIN or ERROR possibly...
+        $qp->finish_continuation unless $pending;
+        return;
+    }
+    
+    my $conn = $qp->connection;
+    my $ip = $conn->remote_ip;
+    $template =~ s/%IP%/$ip/g;
+    $conn->notes('dnsbl', $template) unless $conn->notes('dnsbl');
+    $qp->finish_continuation unless $pending;
+}
 
+sub process_txt_result {
+    my ($qp, $result, $query) = @_;
+    
+    my $pending = $qp->transaction->notes('pending_dns_queries');
+    $qp->transaction->notes('pending_dns_queries', --$pending);
+    
+    warn("Result for TXT $query: $result\n");
+    if ($result !~ /[a-z]/) {
+        # NXDOMAIN or ERROR probably...
+        $qp->finish_continuation unless $pending;
+        return;
+    }
+    
+    my $conn = $qp->connection;
+    $conn->notes('dnsbl', $result) unless $conn->notes('dnsbl');
+    $qp->finish_continuation unless $pending;
 }
 
-sub hook_rcpt {
+sub pickup_handler {
   my ($self, $transaction, $rcpt) = @_;
-  my $connection = $self->qp->connection;
 
   # RBLSMTPD being non-empty means it contains the failure message to return
   if (defined ($ENV{'RBLSMTPD'}) && $ENV{'RBLSMTPD'} ne '') {
     my $result = $ENV{'RBLSMTPD'};
-    my $remote_ip = $connection->remote_ip;
+    my $remote_ip = $self->connection->remote_ip;
     $result =~ s/%IP%/$remote_ip/g;
-    return ($self->{_dnsbl}->{DENY}, 
-	join(" ", $self->qp->config('dnsbl_rejectmsg'), $result));
+    return (DENY, join(" ", $self->config('dnsbl_rejectmsg'), $result));
   }
 
-  my $note = $self->process_sockets;
-  my $whitelist = $connection->notes('whitelisthost');
-  if ( $note ) {
-    if ( $rcpt->user =~ /^(?:postmaster|abuse|mailer-daemon|root)$/i ) {
-      $self->log(LOGWARN, "Don't blacklist special account: ".$rcpt->user);
-    }
-    elsif ( $whitelist ) {
-      $self->log(LOGWARN, "Whitelist overrode blacklist: $whitelist");
-    }
-    elsif ( $connection->relay_client() ) {
-      $self->log(LOGWARN, "Don't blacklist relay/auth clients");
-    }
-    else {
-      return ($self->{_dnsbl}->{DENY}, $note);
-    }
-  }
-  return DECLINED;
-
-}
-
-sub hook_disconnect {
-  my ($self, $transaction) = @_;
-
-  $self->qp->connection->notes('dnsbl_sockets', undef);
-
+  my $note = $self->connection->notes('dnsbl');
+  return (DENY, $note) if $note;
   return DECLINED;
 }
 
@@ -200,19 +125,6 @@ dnsbl - handle DNS BlackList lookups
 Plugin that checks the IP address of the incoming connection against
 a configurable set of RBL services.
 
-=head1 Usage
-
-Add the following line to the config/plugins file:
-
-  dnsbl [disconnect]
-
-If you want to immediately drop the connection (since some blacklisted 
-servers attempt multiple sends per session), add the optional keyword 
-"disconnect" (case insensitive) to the config line.  In most cases, an
-IP address that is listed should not be given the opportunity to begin
-a new transaction, since even the most volatile blacklists will return
-the same answer for a short period of time (the minimum DNS cache period).
-
 =head1 Configuration files
 
 This plugin uses the following configuration files. All of these are optional.

Modified: trunk/plugins/queue/qmail-queue
==============================================================================
--- trunk/plugins/queue/qmail-queue	(original)
+++ trunk/plugins/queue/qmail-queue	Mon Jul 11 12:10:49 2005
@@ -39,12 +39,12 @@ sub hook_queue {
   my ($self, $transaction) = @_;
 
   # these bits inspired by Peter Samuels "qmail-queue wrapper"
-  pipe(MESSAGE_READER, MESSAGE_WRITER) or fault("Could not create message pipe"), exit;
-  pipe(ENVELOPE_READER, ENVELOPE_WRITER) or fault("Could not create envelope pipe"), exit;
+  pipe(MESSAGE_READER, MESSAGE_WRITER) or die("Could not create message pipe");
+  pipe(ENVELOPE_READER, ENVELOPE_WRITER) or die("Could not create envelope pipe");
 
   my $child = fork();
 
-  not defined $child and fault(451, "Could not fork"), exit;
+  not defined $child and die("Could not fork");
 
   if ($child) {
     # Parent
@@ -52,9 +52,13 @@ sub hook_queue {
                 select(ENVELOPE_WRITER); $| = 1;
     select($oldfh);
 
-    close MESSAGE_READER  or fault("close msg reader fault"),exit;
-    close ENVELOPE_READER or fault("close envelope reader fault"), exit;
+    close MESSAGE_READER  or die("close msg reader fault");
+    close ENVELOPE_READER or die("close envelope reader fault");
 
+    # Note - technically there's a race here because if the exec() below
+    # fails and the writes to MESSAGE_WRITER block we get a deadlocked process.
+    # This check to see if(eof(PIPE)) will catch "most" of these problems.
+    die "Message pipe has been closed" if eof(MESSAGE_WRITER);
     $transaction->header->print(\*MESSAGE_WRITER);
     $transaction->body_resetpos;
     while (my $line = $transaction->body_getline) {
@@ -64,6 +68,7 @@ sub hook_queue {
 
     my @rcpt = map { "T" . $_->address } $transaction->recipients;
     my $from = "F".($transaction->sender->address|| "" );
+    die "Envelope pipe has been closed" if eof(ENVELOPE_WRITER);
     print ENVELOPE_WRITER "$from\0", join("\0",@rcpt), "\0\0"
       or return(DECLINED,"Could not print addresses to queue");
     
@@ -104,6 +109,10 @@ sub hook_queue {
 
     my $rc = exec $queue_exec;
 
+    # close the pipe
+    close(MESSAGE_READER);
+    close(MESSAGE_WRITER);
+    
     exit 6; # we'll only get here if the exec fails
   }
 }

Modified: trunk/plugins/require_resolvable_fromhost
==============================================================================
--- trunk/plugins/require_resolvable_fromhost	(original)
+++ trunk/plugins/require_resolvable_fromhost	Mon Jul 11 12:10:49 2005
@@ -1,46 +1,81 @@
-use Net::DNS qw(mx);
+#!/usr/bin/perl
 
-sub hook_mail {
-  my ($self, $transaction, $sender) = @_;
+use Danga::DNS;
 
-  return DECLINED
-        if ($self->qp->connection->notes('whitelistclient'));
-
-  $sender->format ne "<>"
-    and $self->qp->config("require_resolvable_fromhost")
-    and !$self->check_dns($sender->host)
-    and return (DENYSOFT, 
-                ($sender->host
-                   ? "Could not resolve ". $sender->host
-                   : "FQDN required in the envelope sender"));
-  
-   return DECLINED;
+sub register {
+    my ($self) = @_;
+    $self->register_hook("mail", "mail_handler");
+    $self->register_hook("rcpt", "rcpt_handler");
+}
 
+sub mail_handler {
+    my ($self, $transaction, $sender) = @_;
+    
+    $self->transaction->notes('resolvable', 1);
+    return DECLINED if $sender->format eq "<>";
+    return $self->check_dns($sender->host);
 }
 
 
 sub check_dns {
-  my ($self, $host) = @_;
+    my ($self, $host) = @_;
+    
+    # for stuff where we can't even parse a hostname out of the address
+    return DECLINED unless $host;
+    
+    if( $host =~ m/^\[(\d{1,3}\.){3}\d{1,3}\]$/ ) {
+        $self->transaction->notes('resolvable', 1);
+        return DECLINED;
+    }
 
-  # for stuff where we can't even parse a hostname out of the address
-  return 0 unless $host;
+    $self->transaction->notes('pending_dns_queries', 2);
+    my $qp = $self->qp;
+    $self->log(LOGDEBUG, "Checking $host for MX record in the background");
+    Danga::DNS->new(
+        callback => sub { dns_result($qp, @_) },
+        host => $host,
+        type => "MX",
+        client => $qp->input_sock,
+    );
+    $self->log(LOGDEBUG, "Checking $host for A record in the background");
+    Danga::DNS->new(
+        callback => sub { dns_result($qp, @_) },
+        host => $host,
+        client => $qp->input_sock,
+    );
+    return CONTINUATION;
+}
 
-  return 1 if $host =~ m/^\[(\d{1,3}\.){3}\d{1,3}\]$/;
 
-  my $res = new Net::DNS::Resolver;
-  $res->tcp_timeout(30);
-  $res->udp_timeout(30);
-  return 1 if mx($res, $host);
-  my $query = $res->search($host);
-  if ($query) {
-    foreach my $rr ($query->answer) {
-      return 1 if $rr->type eq "A" or $rr->type eq "MX";
+sub dns_result {
+    my ($qp, $result, $query) = @_;
+    
+    my $pending = $qp->transaction->notes('pending_dns_queries');
+    $qp->transaction->notes('pending_dns_queries', --$pending);
+
+    if ($result =~ /^[A-Z]+$/) {
+        # probably an error
+        $qp->log(LOGDEBUG, "DNS error: $result looking up $query");
+    } else {
+        $qp->transaction->notes('resolvable', 1);
+        $qp->log(LOGDEBUG, "DNS lookup $query returned: $result");
     }
-  }
-  else {
-    $self->log(LOGWARN, "$$ query for $host failed: ", $res->errorstring)
-      unless $res->errorstring eq "NXDOMAIN";
-  }
-  return 0;
+
+    $qp->finish_continuation unless $pending;
 }
 
+
+sub rcpt_handler {
+    my ($self, $transaction) = @_;
+    
+    if (!$transaction->notes('resolvable')) {
+        my $sender = $transaction->sender;
+        $self->log(LOGDEBUG, "Could not resolve " .$sender->host) if $sender->host;
+        return (DENYSOFT, 
+		      ($sender->host
+    		   ? "Could not resolve ". $sender->host
+           	   : "FQDN required in the envelope sender"));
+    }
+    
+    return DECLINED;
+}

Modified: trunk/plugins/rhsbl
==============================================================================
--- trunk/plugins/rhsbl	(original)
+++ trunk/plugins/rhsbl	Mon Jul 11 12:10:49 2005
@@ -1,31 +1,39 @@
+#!/usr/bin/perl
 
-sub hook_mail {
+use Danga::DNS;
+
+sub register {
+  my ($self) = @_;
+
+  $self->register_hook('mail', 'mail_handler');
+  $self->register_hook('rcpt', 'rcpt_handler');
+}
+
+sub mail_handler {
   my ($self, $transaction, $sender) = @_;
 
-  my $res = new Net::DNS::Resolver;
-  my $sel = IO::Select->new();
   my %rhsbl_zones_map = ();
 
   # Perform any RHS lookups in the background. We just send the query packets here
   # and pick up any results in the RCPT handler.
   # MTAs gets confused when you reject mail during MAIL FROM:
 
-    my %rhsbl_zones = map { (split /\s+/, $_, 2)[0,1] } $self->qp->config('rhsbl_zones');
+  my %rhsbl_zones = map { (split /\s+/, $_, 2)[0,1] } $self->config('rhsbl_zones');
 
   if ($sender->format ne '<>' and %rhsbl_zones) {
+    my $helo = $self->connection->hello_host;
     push(my @hosts, $sender->host);
-    #my $helo = $self->qp->connection->hello_host;
-    #push(@hosts, $helo) if $helo && $helo ne $sender->host;
+    push(@hosts, $helo) if $helo && $helo ne $sender->host;
     for my $host (@hosts) {
-    for my $rhsbl (keys %rhsbl_zones) {
+      for my $rhsbl (keys %rhsbl_zones) {
         $self->log(LOGDEBUG, "Checking $host.$rhsbl for A record in the background");
-        $sel->add($res->bgsend("$host.$rhsbl"));
-        $rhsbl_zones_map{"$host.$rhsbl"} = $rhsbl_zones{$rhsbl};
+        Danga::DNS->new(
+          callback => sub { $self->process_result($host, $rhsbl_zones{$rhsbl}, @_) },
+          host => "$host.$rhsbl",
+          client => $self->qp->input_sock,
+        );
+      }
     }
-  }
-
-    %{$self->{_rhsbl_zones_map}} = %rhsbl_zones_map;
-    $transaction->notes('rhsbl_sockets', $sel);
   } else {
     $self->log(LOGDEBUG, 'no RHS checks necessary');
   }
@@ -33,80 +41,28 @@ sub hook_mail {
   return DECLINED;
 }
 
-sub hook_rcpt {
-  my ($self, $transaction, $rcpt) = @_;
-  my $host = $transaction->sender->host;
-  my $hello = $self->qp->connection->hello_host;
-
-  my $result = $self->process_sockets;
-  if ($result && defined($self->{_rhsbl_zones_map}{$result})) {
-    if ($result =~ /^$host\./ ) {
-      return (DENY, "Mail from $host rejected because it " . $self->{_rhsbl_zones_map}{$result});
-    } else {
-      return (DENY, "Mail from HELO $hello rejected because it " . $self->{_rhsbl_zones_map}{$result});
+sub process_result {
+    my ($self, $host, $template, $result, $query) = @_;
+    
+    if ($result !~ /^\d+\.\d+\.\d+\.\d+$/) {
+        # NXDOMAIN or error
+        return;
     }
-  }
-  return (DENY, $result) if $result;
-  return DECLINED;
-}
-
-sub process_sockets {
-  my ($self) = @_;
-  my $trans = $self->transaction;
-  my $result = '';
-
-  return $trans->notes('rhsbl') if $trans->notes('rhsbl');
-
-  my $res = new Net::DNS::Resolver;
-  my $sel = $trans->notes('rhsbl_sockets') or return '';
-
-  $self->log(LOGDEBUG, 'waiting for rhsbl dns');
-
-  # don't wait more than 8 seconds here
-  my @ready = $sel->can_read(8);
-
-  $self->log(LOGDEBUG, 'DONE waiting for rhsbl dns, got ' , scalar @ready, ' answers ...') ;
-  return '' unless @ready;
-
-  for my $socket (@ready) {
-    my $query = $res->bgread($socket);
-    $sel->remove($socket);
-    undef $socket;
-
-    if ($query) {
-      foreach my $rr ($query->answer) {
-        $self->log(LOGDEBUG, 'got an ' . $rr->type . ' record ' . $rr->name);
-        if ($rr->type eq 'A') {
-          $result = $rr->name;
-          $self->log(LOGDEBUG, "A record found for $result with IP " . $rr->address);
-          last;
-        }
-      }
-    } else {
-      $self->log(LOGCRIT, "query failed: ", $res->errorstring) unless $res->errorstring eq 'NXDOMAIN';
+    
+    my $tran = $self->transaction;
+    return if $tran->notes('rhsbl');
+    if ($host eq $tran->sender->host) {
+        $tran->notes('rhsbl', "Mail from $host rejected because it $template");
     }
-
-    if ($result) {
-      #kill any other pending I/O
-      $trans->notes('rhsbl_sockets', undef);
-      return $trans->notes('rhsbl', $result);
+    else {
+        $tran->notes('rhsbl', "Mail from HELO $host rejected because it $template");
     }
-  }
-
-  if ($sel->count) {
-    # loop around if we have dns results left
-    return $self->process_sockets();
-  }
-
-  # if there was more to read; then forget it
-  $trans->notes('rhsbl_sockets', undef);
-
-  return $trans->notes('rhsbl', $result);
 }
 
-sub hook_disconnect {
-  my ($self, $transaction) = @_;
+sub rcpt_handler {
+  my ($self, $transaction, $rcpt) = @_;
 
-  $transaction->notes('rhsbl_sockets', undef);
+  my $result = $transaction->notes('rhsbl');
+  return (DENY, $result) if $result;
   return DECLINED;
 }

Modified: trunk/plugins/tls
==============================================================================
--- trunk/plugins/tls	(original)
+++ trunk/plugins/tls	Mon Jul 11 12:10:49 2005
@@ -21,7 +21,7 @@ MAIL FROM onwards.
 
 =cut
 
-use IO::Socket::SSL qw(debug1 debug2 debug3 debug4);
+use IO::Socket::SSL; # qw(debug1 debug2 debug3 debug4);
 
 sub init {
     my ($self, $qp, $cert, $key) = @_;
@@ -38,7 +38,6 @@ sub init {
         SSL_cipher_list => 'HIGH',
         SSL_server => 1
     ) or die "Could not create SSL context: $!";
-    # now extract the password...
     
     $self->ssl_context($ssl_ctx);
 }
@@ -66,31 +65,44 @@ sub hook_unrecognized_command {
     $self->qp->respond (220, "Go ahead with TLS");
     
     eval {
-        my $tlssocket = IO::Socket::SSL->new_from_fd(
-            fileno(STDIN), '+>',
-            SSL_use_cert => 1,
-            SSL_cert_file => $self->tls_cert,
-            SSL_key_file => $self->tls_key,
-            SSL_cipher_list => 'HIGH',
-            SSL_server => 1,
-            SSL_reuse_ctx => $self->ssl_context,
-        ) or die "Could not create SSL socket: $!";
-    
+        my $tlssocket;
+        if ($self->qp->isa('Danga::Socket')) {
+            # high_perf
+            $tlssocket = IO::Socket::SSL->start_SSL($self->qp->sock,
+                SSL_use_cert => 1,
+                SSL_cert_file => $self->tls_cert,
+                SSL_key_file => $self->tls_key,
+                SSL_cipher_list => 'HIGH',
+                SSL_server => 1,
+                SSL_reuse_ctx => $self->ssl_context,
+            ) or die "Could not convert SSL socket: $!";
+        }
+        else {
+            $tlssocket = IO::Socket::SSL->new_from_fd(
+                fileno(STDIN), '+>',
+                SSL_use_cert => 1,
+                SSL_cert_file => $self->tls_cert,
+                SSL_key_file => $self->tls_key,
+                SSL_cipher_list => 'HIGH',
+                SSL_server => 1,
+                SSL_reuse_ctx => $self->ssl_context,
+            ) or die "Could not create SSL socket: $!";
+        }
+        
         my $conn = $self->connection;
         # Create a new connection object with subset of information collected thus far
-        $self->qp->connection(Qpsmtpd::Connection->new(
-           map { $_ => $conn->$_ }
-                qw(
-                    local_ip
-                    local_port
-                    remote_ip
-                    remote_port
-                    remote_host
-                    remote_info
-                ),
-            ));
+        my $newconn = Qpsmtpd::Connection->new();
+        for (qw(local_ip local_port remote_ip remote_port remote_host remote_info)) {
+            $newconn->$_($conn->$_());
+        }
+        $self->qp->connection($newconn);
         $self->qp->reset_transaction;
-        *STDIN = *STDOUT = $self->connection->notes('tls_socket', $tlssocket);
+        if ($self->qp->isa('Danga::Socket')) {
+            $self->connection->notes('tls_socket', $tlssocket);
+        }
+        else {
+            *STDIN = *STDOUT = $self->connection->notes('tls_socket', $tlssocket);
+        }
         $self->connection->notes('tls_enabled', 1);
     };
     if ($@) {
@@ -131,5 +143,6 @@ sub ssl_context {
 sub bad_ssl_hook {
     my ($self, $transaction) = @_;
     return DENY, "Command refused due to lack of security" if $transaction->notes('ssl_failed');
+    return DECLINED;
 }
 *hook_helo = *hook_data = *hook_rcpt = *hook_mail = *hook_auth = \&bad_ssl_hook;

Modified: trunk/qpsmtpd
==============================================================================
--- trunk/qpsmtpd	(original)
+++ trunk/qpsmtpd	Mon Jul 11 12:10:49 2005
@@ -1,30 +1,455 @@
-#!/usr/bin/perl -Tw
-# Copyright (c) 2001 Ask Bjoern Hansen. See the LICENSE file for details.
-# The "command dispatch" system is taken from colobus - http://trainedmonkey.com/colobus/
-#
-# this is designed to be run under tcpserver (http://cr.yp.to/ucspi-tcp.html)
-# or inetd if you're into that sort of thing
-#
-#
-# For more information see http://develooper.com/code/qpsmtpd/
-#
-#
+#!/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';
+}
 
-use lib 'lib';
-use Qpsmtpd::TcpServer;
 use strict;
-$| = 1;
+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;
+
+$|++;
+
+# For debugging
+# $SIG{USR1} = sub { Carp::confess("USR1") };
+
+use Socket qw(SOMAXCONN IPPROTO_TCP SO_KEEPALIVE TCP_NODELAY SOL_SOCKET);
+
+$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 $LineMode    = 0;
+my $PROCS       = 1;
+my $MAXCONN     = 15;           # max simultaneous connections
+my $USER        = 'smtpd';      # user to suid to
+my $MAXCONNIP   = 5;            # max simultaneous connections from one IP
+my $PAUSED      = 0;
+my $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
+ -c, --limit-connections N : limit concurrent connections to N; default 15
+ -u, --user U              : run as a particular user; defualt 'smtpd'
+ -m, --max-from-ip M       : limit connections from a single IP; default 5
+ -f, --forkmode            : fork a child for each connection
+ -j, --procs J             : spawn J processes; default 1
+ -a, --accept K            : accept up to K conns per loop; default 20
+ -h, --help                : this page
+
+NB: -f and -j are mutually exclusive. If -f flag is not used the server uses
+poll() style loops running inside J child processes. Set J to the number of
+CPUs you have at your disposal.
+
+EOT
+    exit(0);
+}
+
+GetOptions(
+    'p|port=i'              => \$PORT,
+    'l|listen-address=s'    => \$LOCALADDR,
+    'j|procs=i'             => \$PROCS,
+    'd|debug+'              => \$DEBUG,
+    'f|forkmode'            => \$LineMode,
+    'c|limit-connections=i' => \$MAXCONN,
+    'm|max-from-ip=i'       => \$MAXCONNIP,
+    'u|user=s'              => \$USER,
+    'a|accept=i'            => \$NUMACCEPT,
+    'h|help'                => \&help,
+    'use-poll'              => \&force_poll,
+) || 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 ($MAXCONN =~ /^(\d+)$/) { $MAXCONN = $1 } else { &help }
+if ($PROCS =~ /^(\d+)$/) { $PROCS = $1 } else { &help }
+if ($NUMACCEPT =~ /^(\d+)$/) { $NUMACCEPT = $1 } else { &help }
+my $_NUMACCEPT = $NUMACCEPT;
+$::LineMode = $LineMode;
+$PROCS = 1 if $LineMode;
+# This is a bit of a hack, but we get to approximate MAXCONN stuff when we
+# have multiple children listening on the same socket.
+$MAXCONN /= $PROCS;
+$MAXCONNIP /= $PROCS;
+
+sub force_poll {
+    $Danga::Socket::HaveEpoll = 0;
+    $Danga::Socket::HaveKQueue = 0;
+}
+
+Danga::Socket::init_poller();
+
+my $POLL = "with " . ($Danga::Socket::HaveEpoll ? "epoll()" : 
+                    $Danga::Socket::HaveKQueue ? "kqueue()" : "poll()");
+
+my $SERVER;
+my $CONFIG_SERVER;
+
+# Code for inetd/tcpserver mode
+if ($ENV{REMOTE_HOST} or $ENV{TCPREMOTEHOST}) {
+    run_as_inetd();
+    exit(0);
+}
+
+my %childstatus = ();
+
+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;
+    Danga::Socket::init_poller();
+}
+
+sub spawn_child {
+    _fork and return;
+    
+    $SIG{CHLD} = "DEFAULT";
+
+    Qpsmtpd::PollServer->OtherFds(fileno($SERVER) => \&accept_handler);
+    Qpsmtpd::PollServer->EventLoop();
+    exit;
+}
+
+sub sig_chld {
+    $SIG{CHLD} = 'IGNORE';
+    while ( (my $child = waitpid(-1,WNOHANG)) > 0) {
+        last unless $child > 0;
+        print "child $child died\n";
+        delete $childstatus{$child};
+    }
+    return if $LineMode;
+    # restart a new child if in poll server mode
+    spawn_child();
+    $SIG{CHLD} = \&sig_chld;
+}
+
+sub HUNTSMAN {
+  $SIG{CHLD} = 'DEFAULT';
+  kill 'INT' => keys %childstatus;
+  exit(0);
+}
+
+sub run_as_inetd {
+    $LineMode = $::LineMode = 1;
+    
+    my $insock = IO::Handle->new_from_fd(0, "r");
+    IO::Handle::blocking($insock, 0);
+
+    my $outsock = IO::Handle->new_from_fd(1, "w");
+    IO::Handle::blocking($outsock, 0);
+
+    my $client = Danga::Client->new($insock);
+
+    my $out = Qpsmtpd::PollServer->new($outsock);
+    $out->load_plugins;
+    $out->init_logger;
+    $out->input_sock($client);
+    $client->push_back_read("Connect\n");
+    # Cause poll/kevent/epoll to end quickly in first iteration
+    Qpsmtpd::PollServer->AddTimer(1, sub { });
+    
+    while (1) {
+        $client->enable_read;
+        my $line = $client->get_line;
+        last if !defined($line);
+        my $output = $out->process_line($line);
+        $out->write($output) if $output;
+    }
+}
+
+sub run_as_server {
+    local $::MAXconn = $MAXCONN;
+    # 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";
+    $) = "";
+    POSIX::setgid($qgid) or
+          die "unable to change gid: $!\n";
+    POSIX::setuid($quid) or
+          die "unable to change uid: $!\n";
+    $> = $quid;
+    
+    ::log(LOGINFO, 'Running as user '.
+        (getpwuid($>) || $>) .
+        ', group '.
+        (getgrgid($)) || $)));
+
+    # Load plugins here
+    my $plugin_loader = Qpsmtpd::SMTP->new();
+    $plugin_loader->load_plugins;
+    
+    if ($PROCS > 1) {
+        $SIG{'CHLD'} = \&sig_chld;
+        my @kids;
+        for (1..$PROCS) {
+            push @kids, spawn_child();
+        }
+        $SIG{INT} = $SIG{TERM} = sub { $SIG{CHLD} = "IGNORE"; kill 2 => @kids; exit };
+        $plugin_loader->log(LOGDEBUG, "Listening on $PORT with $PROCS children $POLL");
+        sleep while (1);
+    }
+    else {
+        if ($LineMode) {
+            $SIG{INT} = $SIG{TERM} = \&HUNTSMAN;
+        }
+        $plugin_loader->log(LOGDEBUG, "Listening on $PORT with single process $POLL" .
+            ($LineMode ? " (forking server)" : ""));
+        Qpsmtpd::PollServer->OtherFds(fileno($SERVER) => \&accept_handler,
+                                      fileno($CONFIG_SERVER) => \&config_handler,
+                                      );
+        while (1) {
+            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;
+}
+
+# Accept all new connections
+sub accept_handler {
+    my $running;
+    if( $LineMode ) {
+        $running = scalar keys %childstatus;
+    }
+    else {
+        my $descriptors = Danga::Client->DescriptorMap;
+        $running = scalar keys %$descriptors;
+    }
+    
+    for (1 .. $NUMACCEPT) {
+        if ($running >= $MAXCONN) { 
+            ::log(LOGINFO,"Too many connections: $running >= $MAXCONN.");
+            return;
+        }
+        $running++;
+        if (! _accept_handler($running)) {
+            # got here because we have too many accepts.
+            $NUMACCEPT = $_NUMACCEPT;
+            return;
+        }
+    }
+    
+    # got here because we have accept's left.
+    # So double the number we accept next time.
+    $NUMACCEPT *= 2;
+}
+
+use Errno qw(EAGAIN EWOULDBLOCK);
+
+sub _accept_handler {
+    my $running = shift;
+
+    my $csock = $SERVER->accept();
+    if (!$csock) {
+        # warn("accept() failed: $!");
+        return;
+        if ($! == EAGAIN || $! == EWOULDBLOCK) {
+            return;
+        }
+        else {
+            warn("accept() failed: $!");
+            return 1;
+        }
+    }
+    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;
+
+    if (!$LineMode) {
+        # multiplex mode
+        my $client = Qpsmtpd::PollServer->new($csock);
+        my $rem_ip = $client->peer_ip_string;
+        
+        if ($PAUSED) {
+            $client->write("451 Sorry, this server is currently paused\r\n");
+            $client->close;
+            return 1;
+        }
+        
+        if ($MAXCONNIP) {
+            my $num_conn = 1; # seed with current value
+    
+            # If we for-loop directly over values %childstatus, a SIGCHLD
+            # can call REAPER and slip $rip out from under us.  Causes
+            # "Use of freed value in iteration" under perl 5.8.4.
+            my $descriptors = Danga::Client->DescriptorMap;
+            my @obj = values %$descriptors;
+            foreach my $obj (@obj) {
+                local $^W;
+                # This is a bit of a slow way to do this. Wish I could cache the method call.
+                ++$num_conn if ($obj->peer_ip_string eq $rem_ip);
+            }
+            
+            if ($num_conn > $MAXCONNIP) {
+                $client->log(LOGINFO,"Too many connections from $rem_ip: "
+                             ."$num_conn > $MAXCONNIP. Denying connection.");
+                $client->write("451 Sorry, too many connections from $rem_ip, try again later\r\n");
+                $client->close;
+                return 1;
+            }
+            $client->log(LOGINFO, "accepted connection $running/$MAXCONN ($num_conn/$MAXCONNIP) from $rem_ip");
+        }
+        
+        $client->push_back_read("Connect\n");
+        $client->watch_read(1);
+        return 1;
+    }
+
+    # fork-per-connection mode
+    my $rem_ip = $csock->sockhost();
+    
+    if ($MAXCONNIP) {
+        my $num_conn = 1; # seed with current value
 
-delete $ENV{ENV};
-$ENV{PATH} = '/bin:/usr/bin:/var/qmail/bin';
+        my @rip = values %childstatus;
+        foreach my $rip (@rip) {
+          ++$num_conn if (defined $rip && $rip eq $rem_ip);
+        }
+        
+        if ($num_conn > $MAXCONNIP) {
+            ::log(LOGINFO,"Too many connections from $rem_ip: "
+                         ."$num_conn > $MAXCONNIP. Denying connection.");
+            print $csock "451 Sorry, too many connections from $rem_ip, try again later\r\n";
+            close $csock;
+            return 1;
+        }
+    }
+    
+    if (my $pid = _fork) {
+        $childstatus{$pid} = $rem_ip;
+        return $csock->close();
+    }
 
-my $qpsmtpd = Qpsmtpd::TcpServer->new();
-$qpsmtpd->start_connection();
-$qpsmtpd->run();
+    $SERVER->close(); # make sure the child doesn't accept() new connections
+    
+    $SIG{$_} = 'DEFAULT' for keys %SIG;
+    
+    my $client = Qpsmtpd::PollServer->new($csock);
+    $client->push_back_read("Connect\n");
+    # Cause poll/kevent/epoll to end quickly in first iteration
+    Qpsmtpd::PollServer->AddTimer(0.1, sub { });
+    
+    while (1) {
+        $client->enable_read;
+        my $line = $client->get_line;
+        last if !defined($line);
+        my $resp = $client->process_line($line);
+        $client->write($resp) if $resp;
+    }
 
-__END__
+    $client->log(LOGDEBUG, "Finished with child %d.\n", fileno($csock))
+        if $DEBUG;
+    $client->close();
 
+    exit;
+}
 
+########################################################################
 
+sub log {
+  my ($level,$message) = @_;
+  # $level not used yet.  this is reimplemented from elsewhere anyway
+  warn("$$ fd:? $message\n");
+}
 
-1;
+sub pause {
+  my ($pause) = @_;
+  $PAUSED = $pause;
+}

Modified: trunk/qpsmtpd-forkserver
==============================================================================
--- trunk/qpsmtpd-forkserver	(original)
+++ trunk/qpsmtpd-forkserver	Mon Jul 11 12:10:49 2005
@@ -11,6 +11,7 @@ use Qpsmtpd::TcpServer;
 use Qpsmtpd::Constants;
 use IO::Socket;
 use IO::Select;
+use Qpsmtpd::PollServer;
 use Socket;
 use Getopt::Long;
 use POSIX qw(:sys_wait_h :errno_h :signal_h);
@@ -24,6 +25,7 @@ my @LOCALADDR;                       # i
 my $USER      = 'smtpd';             # user to suid to
 my $MAXCONNIP = 5;                   # max simultaneous connections from one IP
 my $PID_FILE   = '';                 # file to which server PID will be written
+our $DEBUG = 0;
 
 sub usage {
         print <<"EOT";
@@ -47,6 +49,7 @@ GetOptions('h|help' => \&usage,
            'p|port=i' => \$PORT,
            'u|user=s' => \$USER,
            'pid-file=s' => \$PID_FILE,
+           'd|debug+' => \$DEBUG,
 	  ) || &usage;
 
 # detaint the commandline
@@ -68,6 +71,12 @@ $ENV{PATH} = '/bin:/usr/bin:/var/qmail/b
 my %childstatus = ();
 
 sub REAPER {
+#  foreach my $chld (keys %childstatus) {
+#    if (defined(waitpid($chld, WNOHANG))) {
+#      ::log(LOGINFO,"cleaning up after $chld");
+#      delete $childstatus{$chld};
+#    }
+#  }
   while ( defined(my $chld = waitpid(-1, WNOHANG)) ){
     last unless $chld > 0;
     ::log(LOGINFO,"cleaning up after $chld");
@@ -212,30 +221,22 @@ while (1) {
        ::log(LOGINFO, "Connection Timed Out"); 
        exit; };
   
-    my $localsockaddr = getsockname($client);
-    my ($lport, $laddr) = sockaddr_in($localsockaddr);
-    $ENV{TCPLOCALIP} = inet_ntoa($laddr);
-    # my ($port, $iaddr) = sockaddr_in($hisaddr);
-    $ENV{TCPREMOTEIP} = inet_ntoa($iaddr);
-    $ENV{TCPREMOTEHOST} = gethostbyaddr($iaddr, AF_INET) || "Unknown";
-  
-    # don't do this!
-    #$0 = "qpsmtpd-forkserver: $ENV{TCPREMOTEIP} / $ENV{TCPREMOTEHOST}";
-  
-    ::log(LOGINFO, "Accepted connection $running/$MAXCONN from $ENV{TCPREMOTEIP} / $ENV{TCPREMOTEHOST}");
+    ::log(LOGINFO, "Accepted connection $running/$MAXCONN");
     
-    # dup to STDIN/STDOUT
-    POSIX::dup2(fileno($client), 0);
-    POSIX::dup2(fileno($client), 1);
+    $::LineMode = 1;
     
-    $qpsmtpd->start_connection
-      (
-       local_ip    => $ENV{TCPLOCALIP},
-       local_port  => $lport,
-       remote_ip   => $ENV{TCPREMOTEIP},
-       remote_port => $port,
-      );
-    $qpsmtpd->run();
+    my $qp = Qpsmtpd::PollServer->new($client);
+    $qp->load_plugins;
+    $qp->init_logger;
+    $qp->push_back_read("Connect\n");
+    Qpsmtpd::PollServer->AddTimer(0.1, sub { });
+    while (1) {
+        $qp->enable_read;
+        my $line = $qp->get_line;
+        last if !defined($line);
+        my $output = $qp->process_line($line);
+        $qp->write($output) if $output;
+    }
     
     exit;                                   # child leaves
   }