mirror of
https://git.proxmox.com/git/proxmox-spamassassin
synced 2025-04-28 12:19:37 +00:00
808 lines
28 KiB
Perl
808 lines
28 KiB
Perl
# <@LICENSE>
|
|
# Licensed to the Apache Software Foundation (ASF) under one or more
|
|
# contributor license agreements. See the NOTICE file distributed with
|
|
# this work for additional information regarding copyright ownership.
|
|
# The ASF licenses this file to you under the Apache License, Version 2.0
|
|
# (the "License"); you may not use this file except in compliance with
|
|
# the License. You may obtain a copy of the License at:
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# </@LICENSE>
|
|
|
|
=head1 NAME
|
|
|
|
Mail::SpamAssassin::AsyncLoop - scanner asynchronous event loop
|
|
|
|
=head1 DESCRIPTION
|
|
|
|
An asynchronous event loop used for long-running operations, performed "in the
|
|
background" during the Mail::SpamAssassin::check() scan operation, such as DNS
|
|
blocklist lookups.
|
|
|
|
=head1 METHODS
|
|
|
|
=over 4
|
|
|
|
=cut
|
|
|
|
package Mail::SpamAssassin::AsyncLoop;
|
|
|
|
use strict;
|
|
use warnings;
|
|
# use bytes;
|
|
use re 'taint';
|
|
|
|
use Time::HiRes qw(time);
|
|
|
|
use Mail::SpamAssassin;
|
|
use Mail::SpamAssassin::Logger;
|
|
use Mail::SpamAssassin::Util qw(idn_to_ascii domain_to_search_list);
|
|
|
|
our @ISA = qw();
|
|
|
|
# obtain timer resolution if possible
|
|
our $timer_resolution;
|
|
BEGIN {
|
|
eval {
|
|
$timer_resolution = Time::HiRes->can('clock_getres')
|
|
? Time::HiRes::clock_getres(Time::HiRes::CLOCK_REALTIME())
|
|
: 0.001; # wild guess, assume resolution is better than 1s
|
|
1;
|
|
} or do {
|
|
$timer_resolution = 1; # Perl's builtin timer ticks at one second
|
|
};
|
|
}
|
|
|
|
#############################################################################
|
|
|
|
sub new {
|
|
# called from PerMsgStatus, a new AsyncLoop object is created
|
|
# for each new message processing
|
|
my $class = shift;
|
|
$class = ref($class) || $class;
|
|
|
|
my ($main) = @_;
|
|
my $self = {
|
|
main => $main,
|
|
queries_started => 0,
|
|
queries_completed => 0,
|
|
pending_lookups => { },
|
|
pending_rules => { }, # maintain pending rules list for meta evaluation
|
|
rules_for_key => { }, # record all rules used by a key for logging
|
|
timing_by_query => { },
|
|
all_lookups => { }, # keyed by "rr_type/domain"
|
|
};
|
|
|
|
bless ($self, $class);
|
|
$self;
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $ent = $async-E<gt>bgsend_and_start_lookup($name, $type, $class, $ent, $cb, %options)
|
|
|
|
Launch async DNS lookups. This is the only official method supported for
|
|
plugins since version 4.0.0. Do not use bgsend and start_lookup separately.
|
|
|
|
Merges duplicate queries automatically, only launches one and calls all
|
|
related callbacks on answer.
|
|
|
|
=over 4
|
|
|
|
=item $name (required)
|
|
|
|
Name to query.
|
|
|
|
=item $type (required)
|
|
|
|
Type to query, A, TXT, NS, etc.
|
|
|
|
=item $class (required/deprecated)
|
|
|
|
Deprecated, ignored, set as undef.
|
|
|
|
=item C<$ent> is a required hash reference containing the following items:
|
|
|
|
=over 4
|
|
|
|
=item $ent-E<gt>{rulename} (required)
|
|
|
|
The rulename that started and/or depends on this query. Required for rule
|
|
dependencies to work correctly. Can be a single rulename, or array of
|
|
multiple rulenames.
|
|
|
|
=item $ent-E<gt>{type} (optional)
|
|
|
|
A string, typically one word, used to describe the type of lookup in log
|
|
messages, such as C<DNSBL>, C<URIBL-A>. If not defined, default is value of
|
|
$type.
|
|
|
|
=item $ent-E<gt>{zone} (optional)
|
|
|
|
A zone specification (typically a DNS zone name - e.g. host, domain, or
|
|
RBL) which may be used as a key to look up per-zone settings. No semantics
|
|
on this parameter is imposed by this module. Currently used to fetch
|
|
by-zone timeouts (from rbl_timeout setting). Defaults to $name.
|
|
|
|
=item $ent-E<gt>{timeout_initial} (optional)
|
|
|
|
An initial value of elapsed time for which we are willing to wait for a
|
|
response (time in seconds, floating point value is allowed). When elapsed
|
|
time since a query started exceeds the timeout value and there are no other
|
|
queries to wait for, the query is aborted. The actual timeout value ranges
|
|
from timeout_initial and gradually approaches timeout_min (see next parameter)
|
|
as the number of already completed queries approaches the number of all
|
|
queries started.
|
|
|
|
If a caller does not explicitly provide this parameter or its value is
|
|
undefined, a default initial timeout value is settable by a configuration
|
|
variable rbl_timeout.
|
|
|
|
If a value of the timeout_initial parameter is below timeout_min, the initial
|
|
timeout is set to timeout_min.
|
|
|
|
=item $ent-E<gt>{timeout_min} (optional)
|
|
|
|
A lower bound (in seconds) to which the actual timeout approaches as the
|
|
number of queries completed approaches the number of all queries started.
|
|
Defaults to 0.2 * timeout_initial.
|
|
|
|
=item $ent-E<gt>{key}, $ent-E<gt>{id} (deprecated)
|
|
|
|
Deprecated, ignored, automatically generated since 4.0.0.
|
|
|
|
=item $ent-E<gt>{YOUR_OWN_ITEM}
|
|
|
|
Any other custom values/objects that you want to pass on to the answer
|
|
callback.
|
|
|
|
=back
|
|
|
|
=item $cb (required)
|
|
|
|
Callback function for answer, called as $cb-E<gt>($ent, $pkt). C<$ent> is the
|
|
same object that bgsend_and_start_lookup was called with. C<$pkt> is the
|
|
packet object for the response, Net::DNS:RR objects can be found from
|
|
$pkt-E<gt>answer.
|
|
|
|
=item %options (required)
|
|
|
|
Hash of options. Only supported and required option is master_deadline:
|
|
|
|
master_deadline => $pms->{master_deadline}
|
|
|
|
=back
|
|
|
|
=cut
|
|
|
|
sub start_queue {
|
|
my($self) = @_;
|
|
|
|
$self->{wait_queue} = 1;
|
|
}
|
|
|
|
sub launch_queue {
|
|
my($self) = @_;
|
|
|
|
delete $self->{wait_queue};
|
|
|
|
if ($self->{bgsend_queue}) {
|
|
dbg("async: launching queued lookups");
|
|
foreach (@{$self->{bgsend_queue}}) {
|
|
$self->bgsend_and_start_lookup(@$_);
|
|
}
|
|
delete $self->{bgsend_queue};
|
|
}
|
|
}
|
|
|
|
sub bgsend_and_start_lookup {
|
|
my $self = shift;
|
|
my($domain, $type, $class, $ent, $cb, %options) = @_;
|
|
|
|
return if $self->{main}->{resolver}->{no_resolver};
|
|
|
|
# Waiting for priority -100 to launch?
|
|
if ($self->{wait_queue}) {
|
|
push @{$self->{bgsend_queue}}, [@_];
|
|
dbg("async: DNS priority not reached, queueing lookup: $domain/$type");
|
|
return $ent;
|
|
}
|
|
|
|
if (!defined $ent->{rulename} && !$self->{rulename_warned}++) {
|
|
my($package, $filename, $line) = caller;
|
|
warn "async: bgsend_and_start_lookup called without rulename, ".
|
|
"from $package ($filename) line $line. You are likely using ".
|
|
"a plugin that is not compatible with SpamAssasin 4.0.0.";
|
|
}
|
|
|
|
$domain =~ s/\.+\z//s; # strip trailing dots, these sometimes still sneak in
|
|
$domain = idn_to_ascii($domain);
|
|
|
|
# At this point the $domain should already be encoded to UTF-8 and
|
|
# IDN converted to ASCII-compatible encoding (ACE). Make sure this is
|
|
# really the case in order to be able to catch any leftover omissions.
|
|
if (utf8::is_utf8($domain)) {
|
|
utf8::encode($domain);
|
|
my($package, $filename, $line) = caller;
|
|
info("bgsend_and_start_lookup: Unicode domain name, expected octets: %s, ".
|
|
"called from %s line %d", $domain, $package, $line);
|
|
} elsif ($domain =~ tr/\x00-\x7F//c) { # is not all-ASCII
|
|
my($package, $filename, $line) = caller;
|
|
info("bgsend_and_start_lookup: non-ASCII domain name: %s, ".
|
|
"called from %s line %d", $domain, $package, $line);
|
|
}
|
|
|
|
my $dnskey = uc($type).'/'.lc($domain);
|
|
my $dns_query_info = $self->{all_lookups}{$dnskey};
|
|
|
|
$ent = {} if !$ent;
|
|
$ent->{id} = undef;
|
|
my $key = $ent->{key} = $dnskey;
|
|
$ent->{query_type} = $type;
|
|
$ent->{query_domain} = $domain;
|
|
$ent->{type} = $type if !exists $ent->{type};
|
|
$ent->{zone} = $domain if !exists $ent->{zone};
|
|
$cb = $ent->{completed_callback} if !$cb; # compatibility with SA < 3.4
|
|
|
|
my @rulenames = grep { defined } (ref $ent->{rulename} ?
|
|
@{$ent->{rulename}} : $ent->{rulename});
|
|
|
|
$self->{rules_for_key}->{$key}{$_} = 1 foreach (@rulenames);
|
|
|
|
if ($dns_query_info) { # DNS query already underway or completed
|
|
if ($dns_query_info->{blocked}) {
|
|
dbg("async: blocked by %s: %s, rules: %s", $dns_query_info->{blocked},
|
|
$dnskey, join(", ", @rulenames));
|
|
return;
|
|
}
|
|
my $id = $ent->{id} = $dns_query_info->{id}; # re-use existing query
|
|
return if !defined $id; # presumably some fatal failure
|
|
my $id_tail = $id; $id_tail =~ s{^\d+/IN/}{};
|
|
lc($id_tail) eq lc($dnskey)
|
|
or info("async: unmatched id %s, key=%s", $id, $dnskey);
|
|
|
|
my $pkt = $dns_query_info->{pkt};
|
|
if (!$pkt) { # DNS query underway, still waiting for results
|
|
# just add our query to the existing one
|
|
push(@{$dns_query_info->{applicants}}, [$ent,$cb]);
|
|
$self->{pending_rules}->{$_}{$key} = 1 foreach (@rulenames);
|
|
dbg("async: query %s already underway, adding no.%d, rules: %s",
|
|
$id, scalar @{$dns_query_info->{applicants}},
|
|
join(", ", @rulenames));
|
|
|
|
} else { # DNS query already completed, re-use results
|
|
# answer already known, just do the callback and be done with it
|
|
delete $self->{pending_rules}->{$_}{$key} foreach (@rulenames);
|
|
if (!$cb) {
|
|
dbg("async: query %s already done, re-using for %s, rules: %s",
|
|
$id, $key, join(", ", @rulenames));
|
|
} else {
|
|
dbg("async: query %s already done, re-using for %s, callback, rules: %s",
|
|
$id, $key, join(", ", @rulenames));
|
|
eval {
|
|
$cb->($ent, $pkt); 1;
|
|
} or do {
|
|
chomp $@;
|
|
# resignal if alarm went off
|
|
die "async: (1) $@\n" if $@ =~ /__alarm__ignore__\(.*\)/s;
|
|
warn sprintf("async: query %s completed, callback %s failed: %s\n",
|
|
$id, $key, $@);
|
|
};
|
|
}
|
|
}
|
|
}
|
|
|
|
else { # no existing query, open a new DNS query
|
|
$dns_query_info = $self->{all_lookups}{$dnskey} = {}; # new query needed
|
|
my($id, $blocked, $check_dbrdom);
|
|
# dns_query_restriction
|
|
my $blocked_by = 'dns_query_restriction';
|
|
my $dns_query_blockages = $self->{main}->{conf}->{dns_query_blocked};
|
|
# dns_block_rule
|
|
my $dns_block_domains = $self->{main}->{conf}->{dns_block_rule_domains};
|
|
if ($dns_query_blockages || $dns_block_domains) {
|
|
my $search_list = domain_to_search_list($domain);
|
|
foreach my $parent_domain ((@$search_list, '*')) {
|
|
if ($dns_query_blockages) {
|
|
$blocked = $dns_query_blockages->{$parent_domain};
|
|
last if defined $blocked; # stop at first defined, can be true or false
|
|
}
|
|
if ($parent_domain ne '*' && exists $dns_block_domains->{$parent_domain}) {
|
|
# save for later check.. ps. untainted already
|
|
$check_dbrdom = $dns_block_domains->{$parent_domain};
|
|
}
|
|
}
|
|
}
|
|
if (!$blocked && $check_dbrdom) {
|
|
my $blockfile =
|
|
$self->{main}->sed_path("__global_state_dir__/dnsblock_${check_dbrdom}");
|
|
if (my $mtime = (stat($blockfile))[9]) {
|
|
if (time - $mtime <= $self->{main}->{conf}->{dns_block_time}) {
|
|
$blocked = 1;
|
|
$blocked_by = 'dns_block_rule';
|
|
} else {
|
|
dbg("async: dns_block_rule removing expired $blockfile");
|
|
unlink($blockfile);
|
|
}
|
|
}
|
|
}
|
|
if ($blocked) {
|
|
dbg("async: blocked by %s: %s, rules: %s", $blocked_by, $dnskey,
|
|
join(", ", @rulenames));
|
|
$dns_query_info->{blocked} = $blocked_by;
|
|
} else {
|
|
dbg("async: launching %s, rules: %s", $dnskey, join(", ", @rulenames));
|
|
$id = $self->{main}->{resolver}->bgsend($domain, $type, $class, sub {
|
|
my($pkt, $pkt_id, $timestamp) = @_;
|
|
# this callback sub is called from DnsResolver::poll_responses()
|
|
# dbg("async: in a bgsend_and_start_lookup callback, id %s", $pkt_id);
|
|
if ($pkt_id ne $id) {
|
|
warn "async: mismatched dns id: got $pkt_id, expected $id\n";
|
|
return;
|
|
}
|
|
$self->set_response_packet($pkt_id, $pkt, $ent->{key}, $timestamp);
|
|
$dns_query_info->{pkt} = $pkt;
|
|
my $cb_count = 0;
|
|
foreach my $tuple (@{$dns_query_info->{applicants}}) {
|
|
my($appl_ent, $appl_cb) = @$tuple;
|
|
my @rulenames = grep { defined } (ref $appl_ent->{rulename} ?
|
|
@{$appl_ent->{rulename}} : $appl_ent->{rulename});
|
|
foreach (@rulenames) {
|
|
delete $self->{pending_rules}->{$_}{$appl_ent->{key}};
|
|
}
|
|
if ($appl_cb) {
|
|
dbg("async: calling callback on key %s, rules: %s",
|
|
$key, join(", ", @rulenames));
|
|
$cb_count++;
|
|
eval {
|
|
$appl_cb->($appl_ent, $pkt); 1;
|
|
} or do {
|
|
chomp $@;
|
|
# resignal if alarm went off
|
|
die "async: (2) $@\n" if $@ =~ /__alarm__ignore__\(.*\)/s;
|
|
warn sprintf("async: query %s completed, callback %s failed: %s\n",
|
|
$id, $appl_ent->{key}, $@);
|
|
};
|
|
}
|
|
}
|
|
delete $dns_query_info->{applicants};
|
|
dbg("async: query $id completed, no callbacks run") if !$cb_count;
|
|
});
|
|
}
|
|
return if !defined $id;
|
|
$dns_query_info->{id} = $ent->{id} = $id;
|
|
push(@{$dns_query_info->{applicants}}, [$ent,$cb]);
|
|
$self->{pending_rules}->{$_}{$key} = 1 foreach (@rulenames);
|
|
$self->_start_lookup($ent, $options{master_deadline});
|
|
}
|
|
return $ent;
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $ent = $async-E<gt>start_lookup($ent, $master_deadline)
|
|
|
|
DIRECT USE DEPRECATED since 4.0.0, please use bgsend_and_start_lookup.
|
|
|
|
=cut
|
|
|
|
sub start_lookup {
|
|
my $self = shift;
|
|
|
|
if (!$self->{start_lookup_warned}++) {
|
|
my($package, $filename, $line) = caller;
|
|
warn "async: deprecated start_lookup called, ".
|
|
"from $package ($filename) line $line. You are likely using ".
|
|
"a plugin that is not compatible with SpamAssasin 4.0.0.";
|
|
}
|
|
|
|
return if $self->{main}->{resolver}->{no_resolver};
|
|
$self->_start_lookup(@_);
|
|
}
|
|
|
|
# Internal use not deprecated. :-)
|
|
sub _start_lookup {
|
|
my ($self, $ent, $master_deadline) = @_;
|
|
|
|
my $id = $ent->{id};
|
|
my $key = $ent->{key};
|
|
defined $id && $id ne '' or die "oops, no id";
|
|
$key or die "oops, no key";
|
|
$ent->{type} or die "oops, no type";
|
|
|
|
my $now = time;
|
|
$ent->{start_time} = $now if !defined $ent->{start_time};
|
|
|
|
# are there any applicable per-zone settings?
|
|
my $zone = $ent->{zone};
|
|
my $settings; # a ref to a by-zone or to global settings
|
|
my $conf_by_zone = $self->{main}->{conf}->{by_zone};
|
|
if (defined $zone && $conf_by_zone) {
|
|
# dbg("async: searching for by_zone settings for $zone");
|
|
$zone =~ s/^\.//; $zone =~ s/\.\z//; # strip leading and trailing dot
|
|
for (;;) { # 2.10.example.com, 10.example.com, example.com, com, ''
|
|
if (exists $conf_by_zone->{$zone}) {
|
|
$settings = $conf_by_zone->{$zone};
|
|
last;
|
|
} elsif ($zone eq '') {
|
|
last;
|
|
} else { # strip one level, careful with address literals
|
|
$zone = ($zone =~ /^( (?: [^.] | \[ (?: \\. | [^\]\\] )* \] )* )
|
|
\. (.*) \z/xs) ? $2 : '';
|
|
}
|
|
}
|
|
}
|
|
|
|
dbg("async: applying by_zone settings for %s", $zone) if $settings;
|
|
|
|
my $t_init = $ent->{timeout_initial}; # application-specified has precedence
|
|
$t_init = $settings->{rbl_timeout} if $settings && !defined $t_init;
|
|
$t_init = $self->{main}->{conf}->{rbl_timeout} if !defined $t_init;
|
|
$t_init = 0 if !defined $t_init; # last-resort default, just in case
|
|
|
|
my $t_end = $ent->{timeout_min}; # application-specified has precedence
|
|
$t_end = $settings->{rbl_timeout_min} if $settings && !defined $t_end;
|
|
$t_end = $self->{main}->{conf}->{rbl_timeout_min} if !defined $t_end; # added for bug 7070
|
|
$t_end = 0.2 * $t_init if !defined $t_end;
|
|
$t_end = 0 if $t_end < 0; # just in case
|
|
$t_init = $t_end if $t_init < $t_end;
|
|
|
|
my $clipped_by_master_deadline = 0;
|
|
if (defined $master_deadline) {
|
|
my $time_avail = $master_deadline - time;
|
|
$time_avail = 0.5 if $time_avail < 0.5; # give some slack
|
|
if ($t_init > $time_avail) {
|
|
$t_init = $time_avail; $clipped_by_master_deadline = 1;
|
|
$t_end = $time_avail if $t_end > $time_avail;
|
|
}
|
|
}
|
|
$ent->{timeout_initial} = $t_init;
|
|
$ent->{timeout_min} = $t_end;
|
|
|
|
my @rulenames = grep { defined } (ref $ent->{rulename} ?
|
|
@{$ent->{rulename}} : $ent->{rulename});
|
|
$ent->{display_id} = # identifies entry in debug logging and similar
|
|
join(", ", grep { defined } map { $ent->{$_} } qw(type key));
|
|
|
|
$self->{pending_lookups}->{$key} = $ent;
|
|
|
|
$self->{queries_started}++;
|
|
dbg("async: starting: %s%s (timeout %.1fs, min %.1fs)%s",
|
|
@rulenames ? join(", ", @rulenames).", " : '',
|
|
$ent->{display_id}, $ent->{timeout_initial}, $ent->{timeout_min},
|
|
!$clipped_by_master_deadline ? '' : ', capped by time limit');
|
|
|
|
$ent;
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $ent = $async-E<gt>get_lookup($key)
|
|
|
|
DEPRECATED since 4.0.0. Do not use.
|
|
|
|
=cut
|
|
|
|
sub get_lookup {
|
|
my ($self, $key) = @_;
|
|
warn("async: deprecated get_lookup function used\n");
|
|
return $self->{pending_lookups}->{$key};
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $async-E<gt>log_lookups_timing()
|
|
|
|
Log sorted timing for all completed lookups.
|
|
|
|
=cut
|
|
|
|
sub log_lookups_timing {
|
|
my ($self) = @_;
|
|
my $timings = $self->{timing_by_query};
|
|
for my $key (sort { $timings->{$a} <=> $timings->{$b} } keys %$timings) {
|
|
dbg("async: timing: %.3f %s", $timings->{$key}, $key);
|
|
}
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $alldone = $async-E<gt>complete_lookups()
|
|
|
|
Perform a poll of the pending lookups, to see if any are completed.
|
|
Callbacks on completed queries will be called from poll_responses().
|
|
|
|
If there are no lookups remaining, or if too much time has elapsed since
|
|
any results were returned, C<1> is returned, otherwise C<0>.
|
|
|
|
=cut
|
|
|
|
sub complete_lookups {
|
|
my ($self, $timeout, $allow_aborting_of_expired) = @_;
|
|
my $alldone = 0;
|
|
my $anydone = 0;
|
|
my $allexpired = 1;
|
|
my %typecount;
|
|
|
|
my $pending = $self->{pending_lookups};
|
|
|
|
my $now = time;
|
|
|
|
if (defined $timeout && $timeout > 0 &&
|
|
%$pending && $self->{queries_started} > 0)
|
|
{
|
|
# shrink a 'select' timeout if a caller specified unnecessarily long
|
|
# value beyond the latest deadline of any outstanding request;
|
|
# can save needless wait time (up to 1 second in harvest_dnsbl_queries)
|
|
my $r = $self->{queries_completed} / $self->{queries_started};
|
|
my $r2 = $r * $r; # 0..1
|
|
my $max_deadline;
|
|
while (my($key,$ent) = each %$pending) {
|
|
my $t_init = $ent->{timeout_initial};
|
|
my $dt = $t_init - ($t_init - $ent->{timeout_min}) * $r2;
|
|
my $deadline = $ent->{start_time} + $dt;
|
|
$max_deadline = $deadline if !defined $max_deadline ||
|
|
$deadline > $max_deadline;
|
|
}
|
|
if (defined $max_deadline) {
|
|
# adjust to timer resolution, only deals with 1s and with fine resolution
|
|
$max_deadline = 1 + int $max_deadline
|
|
if $timer_resolution == 1 && $max_deadline > int $max_deadline;
|
|
my $sufficient_timeout = $max_deadline - $now;
|
|
$sufficient_timeout = 0 if $sufficient_timeout < 0;
|
|
if ($timeout > $sufficient_timeout) {
|
|
dbg("async: reducing select timeout from %.1f to %.1f s",
|
|
$timeout, $sufficient_timeout);
|
|
$timeout = $sufficient_timeout;
|
|
}
|
|
}
|
|
}
|
|
|
|
# trap this loop in an eval { } block, as Net::DNS could throw
|
|
# die()s our way; in particular, process_dnsbl_results() has
|
|
# thrown die()s before (bug 3794).
|
|
eval {
|
|
|
|
if (%$pending) { # any outstanding requests still?
|
|
$self->{last_poll_responses_time} = $now;
|
|
my ($nfound, $ncb) = $self->{main}->{resolver}->poll_responses($timeout);
|
|
dbg("async: select found %d responses ready (t.o.=%.1f), did %d callbacks",
|
|
$nfound, $timeout, $ncb);
|
|
}
|
|
$now = time; # capture new timestamp, after possible sleep in 'select'
|
|
|
|
# A callback routine may generate another DNS query, which may insert
|
|
# an entry into the %$pending hash thus invalidating the each() context.
|
|
# So, make sure that callbacks are not called while the each() context
|
|
# is open. [Bug 6937]
|
|
#
|
|
while (my($key,$ent) = each %$pending) {
|
|
my $id = $ent->{id};
|
|
if (exists $self->{finished}->{$id}) {
|
|
delete $self->{finished}->{$id};
|
|
$anydone = 1;
|
|
$ent->{finish_time} = $now if !defined $ent->{finish_time};
|
|
my $elapsed = $ent->{finish_time} - $ent->{start_time};
|
|
my @rulenames = keys %{$self->{rules_for_key}->{$key}};
|
|
dbg("async: completed in %.3f s: %s, rules: %s",
|
|
$elapsed, $ent->{display_id}, join(", ", @rulenames));
|
|
$self->{timing_by_query}->{". $key ($ent->{type})"} += $elapsed;
|
|
$self->{queries_completed}++;
|
|
delete $pending->{$key};
|
|
}
|
|
}
|
|
|
|
if (%$pending) { # still any requests outstanding? are they expired?
|
|
my $r =
|
|
!$allow_aborting_of_expired || !$self->{queries_started} ? 1.0
|
|
: $self->{queries_completed} / $self->{queries_started};
|
|
my $r2 = $r * $r; # 0..1
|
|
while (my($key,$ent) = each %$pending) {
|
|
$typecount{$ent->{type}}++;
|
|
my $t_init = $ent->{timeout_initial};
|
|
my $dt = $t_init - ($t_init - $ent->{timeout_min}) * $r2;
|
|
# adjust to timer resolution, only deals with 1s and fine resolution
|
|
$dt = 1 + int $dt if $timer_resolution == 1 && $dt > int $dt;
|
|
$allexpired = 0 if $now <= $ent->{start_time} + $dt;
|
|
}
|
|
}
|
|
|
|
# ensure we don't get stuck if a request gets lost in the ether.
|
|
if (! %$pending) {
|
|
$alldone = 1;
|
|
}
|
|
elsif ($allexpired && $allow_aborting_of_expired) {
|
|
# avoid looping forever if we haven't got all results.
|
|
dbg("async: escaping: lost or timed out requests or responses");
|
|
$self->abort_remaining_lookups();
|
|
$alldone = 1;
|
|
}
|
|
else {
|
|
dbg("async: queries still pending: %s%s",
|
|
join (' ', map { "$_=$typecount{$_}" } sort keys %typecount),
|
|
$allexpired ? ', all expired' : '');
|
|
$alldone = 0;
|
|
}
|
|
1;
|
|
|
|
} or do {
|
|
my $eval_stat = $@ ne '' ? $@ : "errno=$!"; chomp $eval_stat;
|
|
# resignal if alarm went off
|
|
die "async: (3) $eval_stat\n" if $eval_stat =~ /__alarm__ignore__\(.*\)/s;
|
|
dbg("async: caught complete_lookups death, aborting: %s", $eval_stat);
|
|
$alldone = 1; # abort remaining
|
|
};
|
|
|
|
return wantarray ? ($alldone,$anydone) : $alldone;
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $async-E<gt>abort_remaining_lookups()
|
|
|
|
Abort any remaining lookups.
|
|
|
|
=cut
|
|
|
|
sub abort_remaining_lookups {
|
|
my ($self) = @_;
|
|
|
|
my $pending = $self->{pending_lookups};
|
|
my $foundcnt = 0;
|
|
my $now = time;
|
|
|
|
$self->{pending_rules} = {};
|
|
|
|
while (my($key,$ent) = each %$pending) {
|
|
my $dur = $now - $ent->{start_time};
|
|
my @rulenames = keys %{$self->{rules_for_key}->{$key}};
|
|
my $msg = sprintf( "async: aborting after %.3f s, %s: %s, rules: %s",
|
|
$dur,
|
|
(defined $ent->{timeout_initial} &&
|
|
$now > $ent->{start_time} + $ent->{timeout_initial}
|
|
? 'past original deadline' : 'deadline shrunk'),
|
|
$ent->{display_id}, join(", ", @rulenames) );
|
|
$dur > 1 ? info($msg) : dbg($msg);
|
|
$foundcnt++;
|
|
$self->{timing_by_query}->{"X $key"} = $dur;
|
|
$ent->{finish_time} = $now if !defined $ent->{finish_time};
|
|
delete $pending->{$key};
|
|
}
|
|
|
|
# call any remaining callbacks, indicating the query has been aborted
|
|
#
|
|
my $all_lookups_ref = $self->{all_lookups};
|
|
foreach my $dnskey (keys %$all_lookups_ref) {
|
|
my $dns_query_info = $all_lookups_ref->{$dnskey};
|
|
my $cb_count = 0;
|
|
foreach my $tuple (@{$dns_query_info->{applicants}}) {
|
|
my($ent, $cb) = @$tuple;
|
|
if ($cb) {
|
|
my @rulenames = grep { defined } (ref $ent->{rulename} ?
|
|
@{$ent->{rulename}} : $ent->{rulename});
|
|
dbg("async: calling callback/abort on key %s, rules: %s", $dnskey,
|
|
join(", ", @rulenames));
|
|
$cb_count++;
|
|
eval {
|
|
$cb->($ent, undef); 1;
|
|
} or do {
|
|
chomp $@;
|
|
# resignal if alarm went off
|
|
die "async: (2) $@\n" if $@ =~ /__alarm__ignore__\(.*\)/s;
|
|
warn sprintf("async: query %s aborted, callback %s failed: %s\n",
|
|
$dnskey, $ent->{key}, $@);
|
|
};
|
|
}
|
|
dbg("async: query $dnskey aborted, no callbacks run") if !$cb_count;
|
|
}
|
|
delete $dns_query_info->{applicants};
|
|
}
|
|
|
|
dbg("async: aborted %d remaining lookups", $foundcnt) if $foundcnt > 0;
|
|
delete $self->{last_poll_responses_time};
|
|
$self->{main}->{resolver}->bgabort();
|
|
1;
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $async-E<gt>set_response_packet($id, $pkt, $key, $timestamp)
|
|
|
|
For internal use, do not call from plugins.
|
|
|
|
Register a "response packet" for a given query. C<$id> is the ID for the
|
|
query, and must match the C<id> supplied in C<start_lookup()>. C<$pkt> is the
|
|
packet object for the response. A parameter C<$key> identifies an entry in a
|
|
hash %{$self-E<gt>{pending_lookups}} where the object which spawned this query can
|
|
be found, and through which further information about the query is accessible.
|
|
|
|
C<$pkt> may be undef, indicating that no response packet is available, but a
|
|
query has completed (e.g. was aborted or dismissed) and is no longer "pending".
|
|
|
|
The DNS resolver's response packet C<$pkt> will be made available to a callback
|
|
subroutine through its argument as well as in C<$ent-E<gt>{response_packet}>.
|
|
|
|
=cut
|
|
|
|
sub set_response_packet {
|
|
my ($self, $id, $pkt, $key, $timestamp) = @_;
|
|
$self->{finished}->{$id} = 1; # only key existence matters, any value
|
|
$timestamp = time if !defined $timestamp;
|
|
my $pending = $self->{pending_lookups};
|
|
if (!defined $key) { # backward compatibility with 3.2.3 and older plugins
|
|
# a third-party plugin did not provide $key in a call, search for it:
|
|
if ($id eq $pending->{$id}->{id}) { # I feel lucky, key==id ?
|
|
$key = $id;
|
|
} else { # then again, maybe not, be more systematic
|
|
for my $tkey (keys %$pending) {
|
|
if ($id eq $pending->{$tkey}->{id}) { $key = $tkey; last }
|
|
}
|
|
}
|
|
dbg("async: got response on id $id, search found key $key");
|
|
}
|
|
if (!defined $key) {
|
|
info("async: no key, response packet not remembered, id $id");
|
|
} else {
|
|
my $ent = $pending->{$key};
|
|
my $ent_id = $ent->{id};
|
|
if (!defined $ent_id) {
|
|
# should not happen, troubleshooting
|
|
info("async: ignoring response, id %s, ent_id is undef: %s",
|
|
$id, join(', ', %$ent));
|
|
} elsif ($id ne $ent_id) {
|
|
info("async: ignoring response, mismatched id $id, expected $ent_id");
|
|
} else {
|
|
$ent->{finish_time} = $timestamp;
|
|
$ent->{response_packet} = $pkt;
|
|
}
|
|
}
|
|
1;
|
|
}
|
|
|
|
=item $async-E<gt>report_id_complete($id,$key,$key,$timestamp)
|
|
|
|
DEPRECATED since 4.0.0. Do not use.
|
|
|
|
Legacy. Equivalent to $self-E<gt>set_response_packet($id,undef,$key,$timestamp),
|
|
i.e. providing undef as a response packet. Register that a query has
|
|
completed and is no longer "pending". C<$id> is the ID for the query,
|
|
and must match the C<id> supplied in C<start_lookup()>.
|
|
|
|
One or the other of C<set_response_packet()> or C<report_id_complete()>
|
|
should be called, but not both.
|
|
|
|
=cut
|
|
|
|
sub report_id_complete {
|
|
my ($self, $id, $key, $timestamp) = @_;
|
|
$self->set_response_packet($id, undef, $key, $timestamp);
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
=item $time = $async-E<gt>last_poll_responses_time()
|
|
|
|
Get the time of the last call to C<poll_responses()> (which is called
|
|
from C<complete_lookups()>. If C<poll_responses()> was never called or
|
|
C<abort_remaining_lookups()> has been called C<last_poll_responses_time()>
|
|
will return undef.
|
|
|
|
=cut
|
|
|
|
sub last_poll_responses_time {
|
|
my ($self) = @_;
|
|
return $self->{last_poll_responses_time};
|
|
}
|
|
|
|
1;
|
|
|
|
=back
|
|
|
|
=cut
|