byterazor/qpsmtpd
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support all resolvers in resolv.conf, and issue retries on errors the same

Browse Source 
way gethostbyname() does.


git-svn-id: https://svn.perl.org/qpsmtpd/trunk@555 958fd67b-6ff1-0310-b445-bb7760255be9
This commit is contained in:
Matt Sergeant 2005-10-26 19:09:04 +00:00
parent 22fef51c19
commit 0a397e74a9
1 changed files with 179 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

256
lib/Danga/DNS/Resolver.pm
View File

@ -3,7 +3,7 @@
package Danga::DNS::Resolver; package Danga::DNS::Resolver;
use base qw(Danga::Socket); use base qw(Danga::Socket);
use fields qw(res dst id_to_asker id_to_query timeout cache cache_timeout); use fields qw(res dst cache cache_timeout queries);
use Net::DNS; use Net::DNS;
use Socket; use Socket;
@ -30,16 +30,16 @@ sub new {
) || die "Cannot create socket: $!"; ) || die "Cannot create socket: $!";
IO::Handle::blocking($sock, 0); IO::Handle::blocking($sock, 0);
trace(2, "Using nameserver $res->{nameservers}->[0]:$res->{port}\n"); $self->{dst} = [];
my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton($res->{'nameservers'}->[0]));
#my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton('127.0.0.1')); foreach my $ns (@{ $res->{nameservers} }) {
#my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton('10.2.1.20')); trace(2, "Using nameserver $ns:$res->{port}\n");
my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton($ns));
push @{$self->{dst}}, $dst_sockaddr;
}
$self->{res} = $res; $self->{res} = $res;
$self->{dst} = $dst_sockaddr; $self->{queries} = {};
$self->{id_to_asker} = {};
$self->{id_to_query} = {};
$self->{timeout} = {};
$self->{cache} = {}; $self->{cache} = {};
$self->{cache_timeout} = {}; $self->{cache_timeout} = {};
@ -52,10 +52,17 @@ sub new {
return $self; return $self;
} }
sub ns {
my Danga::DNS::Resolver $self = shift;
my $index = shift;
return if $index > $#{$self->{dst}};
return $self->{dst}->[$index];
}
sub pending { sub pending {
my Danga::DNS::Resolver $self = shift; my Danga::DNS::Resolver $self = shift;
return keys(%{$self->{id_to_asker}}); return keys(%{$self->{queries}});
} }
sub _query { sub _query {
@ -73,20 +80,14 @@ sub _query {
} }
my $packet = $self->{res}->make_query_packet($host, $type); my $packet = $self->{res}->make_query_packet($host, $type);
my $packet_data = $packet->data; my $packet_data = $packet->data;
my $id = $packet->header->id;
my $h = $packet->header; my $query = Danga::DNS::Resolver::Query->new(
my $id = $h->id; $self, $asker, $host, $type, $now, $id, $packet_data,
) or return;
if (!$self->sock->send($packet_data, 0, $self->{dst})) { $self->{queries}->{$id} = $query;
return;
}
trace(2, "Query: $host ($id)\n");
$self->{id_to_asker}->{$id} = $asker;
$self->{id_to_query}->{$id} = $host;
$self->{timeout}->{$id} = $now;
return 1; return 1;
} }
@ -97,15 +98,12 @@ sub query_txt {
my $now = time(); my $now = time();
trace(2, "[" . keys(%{$self->{id_to_asker}}) . "] trying to resolve TXT: @hosts\n"); trace(2, "trying to resolve TXT: @hosts\n");
foreach my $host (@hosts) { foreach my $host (@hosts) {
$self->_query($asker, $host, 'TXT', $now) || return; $self->_query($asker, $host, 'TXT', $now) || return;
} }
#print "+Pending queries: " . keys(%{$self->{id_to_asker}}) .
# " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
return 1; return 1;
} }
@ -115,15 +113,12 @@ sub query_mx {
my $now = time(); my $now = time();
trace(2, "[" . keys(%{$self->{id_to_asker}}) . "] trying to resolve MX: @hosts\n"); trace(2, "trying to resolve MX: @hosts\n");
foreach my $host (@hosts) { foreach my $host (@hosts) {
$self->_query($asker, $host, 'MX', $now) || return; $self->_query($asker, $host, 'MX', $now) || return;
} }
#print "+Pending queries: " . keys(%{$self->{id_to_asker}}) .
# " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
return 1; return 1;
} }
@ -133,15 +128,12 @@ sub query {
my $now = time(); my $now = time();
trace(2, "[" . keys(%{$self->{id_to_asker}}) . "] trying to resolve A/PTR: @hosts\n"); trace(2, "trying to resolve A/PTR: @hosts\n");
foreach my $host (@hosts) { foreach my $host (@hosts) {
$self->_query($asker, $host, 'A', $now) || return; $self->_query($asker, $host, 'A', $now) || return;
} }
#print "+Pending queries: " . keys(%{$self->{id_to_asker}}) .
# " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
return 1; return 1;
} }
@ -154,20 +146,20 @@ sub _do_cleanup {
my $idle = $self->max_idle_time; my $idle = $self->max_idle_time;
my @to_delete; my @to_delete;
while (my ($id, $t) = each(%{$self->{timeout}})) { while (my ($id, $obj) = each(%{$self->{queries}})) {
if ($t < ($now - $idle)) { if ($obj->{timeout} < ($now - $idle)) {
push @to_delete, $id; push @to_delete, $id;
} }
} }
foreach my $id (@to_delete) { foreach my $id (@to_delete) {
delete $self->{timeout}{$id}; my $query = delete $self->{queries}{$id};
my $asker = delete $self->{id_to_asker}{$id}; $query->timeout() and next;
my $query = delete $self->{id_to_query}{$id}; # add back in if timeout caused us to loop to next server
$asker->run_callback("NXDOMAIN", $query); $self->{queries}->{$id} = $query;
} }
foreach my $type ('A', 'TXT') { foreach my $type ('A', 'TXT', 'MX') {
@to_delete = (); @to_delete = ();
while (my ($query, $t) = each(%{$self->{cache_timeout}{$type}})) { while (my ($query, $t) = each(%{$self->{cache_timeout}{$type}})) {
@ -199,17 +191,14 @@ sub event_read {
my $header = $packet->header; my $header = $packet->header;
my $id = $header->id; my $id = $header->id;
my $asker = delete $self->{id_to_asker}->{$id}; my $qobj = delete $self->{queries}->{$id};
my $query = delete $self->{id_to_query}->{$id}; if (!$qobj) {
delete $self->{timeout}{$id}; trace(1, "No query for id: $id\n");
#print "-Pending queries: " . keys(%{$self->{id_to_asker}}) .
# " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
if (!$asker) {
trace(1, "No asker for id: $id\n");
return; return;
} }
my $query = $qobj->{host};
my $now = time(); my $now = time();
my @questions = $packet->question; my @questions = $packet->question;
#print STDERR "response to ", $questions[0]->string, "\n"; #print STDERR "response to ", $questions[0]->string, "\n";
@ -217,61 +206,64 @@ sub event_read {
# my $q = shift @questions; # my $q = shift @questions;
if ($rr->type eq "PTR") { if ($rr->type eq "PTR") {
my $rdns = $rr->ptrdname; my $rdns = $rr->ptrdname;
if ($query) { # NB: Cached as an "A" lookup as there's no overlap and they
# NB: Cached as an "A" lookup as there's no overlap and they # go through the same query() function above
# go through the same query() function above $self->{cache}{A}{$query} = $rdns;
$self->{cache}{A}{$query} = $rdns; # $self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
$self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long $self->{cache_timeout}{A}{$query} = $now + $rr->ttl;
} $qobj->run_callback($rdns);
$asker->run_callback($rdns, $query);
} }
elsif ($rr->type eq "A") { elsif ($rr->type eq "A") {
my $ip = $rr->address; my $ip = $rr->address;
if ($query) { $self->{cache}{A}{$query} = $ip;
$self->{cache}{A}{$query} = $ip; # $self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
$self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long $self->{cache_timeout}{A}{$query} = $now + $rr->ttl;
} $qobj->run_callback($ip);
$asker->run_callback($ip, $query);
} }
elsif ($rr->type eq "TXT") { elsif ($rr->type eq "TXT") {
my $txt = $rr->txtdata; my $txt = $rr->txtdata;
if ($query) { $self->{cache}{TXT}{$query} = $txt;
$self->{cache}{TXT}{$query} = $txt; # $self->{cache_timeout}{TXT}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
$self->{cache_timeout}{TXT}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long $self->{cache_timeout}{TXT}{$query} = $now + $rr->ttl;
} $qobj->run_callback($txt);
$asker->run_callback($txt, $query); }
elsif ($rr->type eq "MX") {
my $host = $rr->exchange;
my $preference = $rr->preference;
$self->{cache}{MX}{$query} = [$host, $preference];
$self->{cache_timeout}{MX}{$query} = $now + $rr->ttl;
$qobj->run_callback([$host, $preference]);
} }
else { else {
# came back, but not a PTR or A record # came back, but not a PTR or A record
$asker->run_callback("unknown", $query); $qobj->run_callback("UNKNOWN");
} }
$answers++; $answers++;
} }
if (!$answers) { if (!$answers) {
if ($err eq "NXDOMAIN") { if ($err eq "NXDOMAIN") {
# trace("found => NXDOMAIN\n"); # trace("found => NXDOMAIN\n");
$asker->run_callback("NXDOMAIN", $query); $qobj->run_callback("NXDOMAIN");
} }
elsif ($err eq "SERVFAIL") { elsif ($err eq "SERVFAIL") {
# try again??? # try again???
print "SERVFAIL looking for $query (Pending: " . keys(%{$self->{id_to_asker}}) . ")\n"; print "SERVFAIL looking for $query\n";
#$self->query($asker, $query); #$self->query($asker, $query);
$asker->run_callback($err, $query); $qobj->error($err) and next;
#$self->{id_to_asker}->{$id} = $asker; # add back in if error() resulted in query being re-issued
#$self->{id_to_query}->{$id} = $query; $self->{queries}->{$id} = $qobj;
#$self->{timeout}{$id} = time();
} }
elsif ($err eq "NOERROR") { elsif ($err eq "NOERROR") {
$asker->run_callback($err, $query); $qobj->run_callback($err);
} }
elsif($err) { elsif($err) {
print("error: $err\n"); print("error: $err\n");
$asker->run_callback($err, $query); $qobj->error($err) and next;
$self->{queries}->{$id} = $qobj;
} }
else { else {
# trace("no answers\n"); # trace("no answers\n");
$asker->run_callback("NXDOMAIN", $query); $qobj->run_callback("NOANSWER");
} }
} }
} }
@ -286,6 +278,116 @@ sub close {
# confess "Danga::DNS::Resolver socket should never be closed!"; # confess "Danga::DNS::Resolver socket should never be closed!";
} }
package Danga::DNS::Resolver::Query;
use constant MAX_QUERIES => 10;
sub trace {
my $level = shift;
print ("$::DEBUG/$level [$$] dns lookup: @_") if $::DEBUG >= $level;
}
sub new {
my ($class, $res, $asker, $host, $type, $now, $id, $data) = @_;
my $self = {
resolver => $res,
asker => $asker,
host => $host,
type => $type,
timeout => $now,
id => $id,
data => $data,
repeat => 2, # number of retries
ns => 0,
nqueries => 0,
};
trace(2, "NS Query: $host ($id)\n");
bless $self, $class;
$self->send_query || return;
return $self;
}
#sub DESTROY {
# my $self = shift;
# trace(2, "DESTROY $self\n");
#}
sub timeout {
my $self = shift;
trace(2, "NS Query timeout. Trying next host\n");
if ($self->send_query) {
# had another NS to send to, reset timeout
$self->{timeout} = time();
return;
}
# can we loop/repeat?
if (($self->{nqueries} <= MAX_QUERIES) &&
($self->{repeat} > 1))
{
trace(2, "NS Query timeout. Next host failed. Trying loop\n");
$self->{repeat}--;
$self->{ns} = 0;
return $self->timeout();
}
trace(2, "NS Query timeout. All failed. Running callback(TIMEOUT)\n");
# otherwise we really must timeout.
$self->run_callback("TIMEOUT");
return 1;
}
sub error {
my ($self, $error) = @_;
trace(2, "NS Query error. Trying next host\n");
if ($self->send_query) {
# had another NS to send to, reset timeout
$self->{timeout} = time();
return;
}
# can we loop/repeat?
if (($self->{nqueries} <= MAX_QUERIES) &&
($self->{repeat} > 1))
{
trace(2, "NS Query error. Next host failed. Trying loop\n");
$self->{repeat}--;
$self->{ns} = 0;
return $self->error($error);
}
trace(2, "NS Query error. All failed. Running callback($error)\n");
# otherwise we really must timeout.
$self->run_callback($error);
return 1;
}
sub run_callback {
my ($self, $response) = @_;
trace(2, "NS Query callback($self->{host} = $response\n");
$self->{asker}->run_callback($response, $self->{host});
}
sub send_query {
my ($self) = @_;
my $dst = $self->{resolver}->ns($self->{ns}++);
return unless defined $dst;
if (!$self->{resolver}->sock->send($self->{data}, 0, $dst)) {
return;
}
$self->{nqueries}++;
return 1;
}
1; 1;
=head1 NAME =head1 NAME