Asynchronous NBTstat

Netdisco/Changes

@@ -1,3 +1,9 @@
+2.028006
+
+  [ENHANCEMENTS]
+
+  * Asynchronous NBTstat
+
 2.028005 - 2014-07-17
 
   [BUG FIXES]

Netdisco/lib/App/Netdisco/AnyEvent/Nbtstat.pm

@@ -0,0 +1,277 @@
+package App::Netdisco::AnyEvent::Nbtstat;
+
+use Socket qw(AF_INET SOCK_DGRAM inet_aton sockaddr_in);
+use List::Util ();
+use Carp       ();
+
+use AnyEvent (); BEGIN { AnyEvent::common_sense }
+use AnyEvent::Util ();
+
+sub new {
+    my ( $class, %args ) = @_;
+
+    my $interval = $args{interval};
+    # This default should generate ~ 50 requests per second
+    $interval = 0.2 unless defined $interval;
+
+    my $timeout = $args{timeout};
+
+    # Timeout should be 250ms according to RFC1002, but we're going to double
+    $timeout = 0.5 unless defined $timeout;
+
+    my $self = bless { interval => $interval, timeout => $timeout, %args },
+        $class;
+
+    Scalar::Util::weaken( my $wself = $self );
+
+    socket my $fh4, AF_INET, Socket::SOCK_DGRAM(), 0
+        or Carp::croak "Unable to create socket : $!";
+
+    AnyEvent::Util::fh_nonblocking $fh4, 1;
+    $self->{fh4} = $fh4;
+    $self->{rw4} = AE::io $fh4, 0, sub {
+        if ( my $peer = recv $fh4, my $resp, 2048, 0 ) {
+            $wself->_on_read( $resp, $peer );
+        }
+    };
+
+    # Nbtstat tasks
+    $self->{_tasks}     = {};
+
+    return $self;
+}
+
+sub interval { @_ > 1 ? $_[0]->{interval} = $_[1] : $_[0]->{interval} }
+
+sub timeout { @_ > 1 ? $_[0]->{timeout} = $_[1] : $_[0]->{timeout} }
+
+sub nbtstat {
+    my ( $self, $host, $cb ) = @_;
+
+    my $ip   = inet_aton($host);
+    my $port = 137;
+
+    my $request = {
+        host        => $host,
+        results     => {},
+        cb          => $cb,
+        destination => scalar sockaddr_in( $port, $ip ),
+    };
+
+    $self->{_tasks}{ $request->{destination} } = $request;
+
+    my $delay = $self->interval * scalar keys $self->{_tasks};
+
+    # There's probably a better way to throttle the sends
+    # but this will work for now since we currently don't support retries
+    my $w; $w = AE::timer $delay, 0, sub {
+        undef $w;
+        $self->_send_request($request);
+    };
+
+    return $self;
+}
+
+sub _on_read {
+    my ( $self, $resp, $peer ) = @_;
+
+    ($resp) = $resp =~ /^(.*)$/s
+        if AnyEvent::TAINT && $self->{untaint};
+
+    # Find our task
+    my $request = $self->{_tasks}{$peer};
+
+    return unless $request;
+
+    $self->_store_result( $request, 'OK', $resp );
+
+    return;
+}
+
+sub _store_result {
+    my ( $self, $request, $status, $resp ) = @_;
+
+    my $results = $request->{results};
+
+    my @rr          = ();
+    my $mac_address = "";
+
+    if ( $status eq 'OK' && length($resp) > 56 ) {
+        my $num_names = unpack( "C", substr( $resp, 56 ) );
+        my $name_data = substr( $resp, 57 );
+
+        for ( my $i = 0; $i < $num_names; $i++ ) {
+            my $rr_data = substr( $name_data, 18 * $i, 18 );
+            push @rr, _decode_rr($rr_data);
+        }
+
+        $mac_address = join "-",
+            map { sprintf "%02X", $_ }
+            unpack( "C*", substr( $name_data, 18 * $num_names, 6 ) );
+        $results = {
+            'status'      => 'OK',
+            'names'       => \@rr,
+            'mac_address' => $mac_address
+        };
+    }
+    elsif ( $status eq 'OK' ) {
+        $results = { 'status' => 'SHORT' };
+    }
+    else {
+        $results = { 'status' => $status };
+    }
+
+    # Clear request specific data
+    delete $request->{timer};
+
+    # Cleanup
+    delete $self->{_tasks}{ $request->{destination} };
+
+    # Done
+    $request->{cb}->($results);
+
+    undef $request;
+
+    return;
+}
+
+sub _send_request {
+    my ( $self, $request ) = @_;
+
+    my $msg = "";
+    # We use process id as identifier field, since don't have a need to
+    # unique responses beyond host / port queried 
+    $msg .= pack( "n*", $$, 0, 1, 0, 0, 0 );
+    $msg .= _encode_name( "*", "\x00", 0 );
+    $msg .= pack( "n*", 0x21, 0x0001 );
+
+    $request->{start} = time;
+
+    $request->{timer} = AE::timer $self->timeout, 0, sub {
+        $self->_store_result( $request, 'TIMEOUT' );
+    };
+
+    my $fh = $self->{fh4};
+
+    send $fh, $msg, 0, $request->{destination}
+        or $self->_store_result( $request, 'ERROR' );
+
+    return;
+}
+
+sub _encode_name {
+    my $name   = uc(shift);
+    my $pad    = shift || "\x20";
+    my $suffix = shift || 0x00;
+
+    $name .= $pad x ( 16 - length($name) );
+    substr( $name, 15, 1, chr( $suffix & 0xFF ) );
+
+    my $encoded_name = "";
+    for my $c ( unpack( "C16", $name ) ) {
+        $encoded_name .= chr( ord('A') + ( ( $c & 0xF0 ) >> 4 ) );
+        $encoded_name .= chr( ord('A') + ( $c & 0xF ) );
+    }
+
+    # Note that the _encode_name function doesn't add any scope,
+    # nor does it calculate the length (32), it just prefixes it
+    return "\x20" . $encoded_name . "\x00";
+}
+
+sub _decode_rr {
+    my $rr_data = shift;
+
+    my @nodetypes = qw/B-node P-node M-node H-node/;
+    my ( $name, $suffix, $flags ) = unpack( "a15Cn", $rr_data );
+    $name =~ tr/\x00-\x19/\./;    # replace ctrl chars with "."
+    $name =~ s/\s+//g;
+
+    my $rr = {};
+    $rr->{'name'}   = $name;
+    $rr->{'suffix'} = $suffix;
+    $rr->{'G'}      = ( $flags & 2**15 ) ? "GROUP" : "UNIQUE";
+    $rr->{'ONT'}    = $nodetypes[ ( $flags >> 13 ) & 3 ];
+    $rr->{'DRG'}    = ( $flags & 2**12 ) ? "Deregistering" : "Registered";
+    $rr->{'CNF'}    = ( $flags & 2**11 ) ? "Conflict" : "";
+    $rr->{'ACT'}    = ( $flags & 2**10 ) ? "Active" : "Inactive";
+    $rr->{'PRM'}    = ( $flags & 2**9 ) ? "Permanent" : "";
+
+    return $rr;
+}
+
+1;
+__END__
+
+=head1 NAME
+
+App::Netdisco::AnyEvent::Nbtstat - Request NetBIOS node status with AnyEvent
+
+=head1 SYNOPSIS
+
+    use App::Netdisco::AnyEvent::Nbtstat;;
+
+    my $request = App::Netdisco::AnyEvent::Nbtstat->new();
+
+    my $cv = AE::cv;
+
+    $request->nbtstat(
+        '127.0.0.1',
+        sub {
+            my $result = shift;
+            print "MAC: ", $result->{'mac_address'} || '', " ";
+            print "Status: ", $result->{'status'}, "\n";
+            printf '%3s %-18s %4s %-18s', '', 'Name', '', 'Type'
+                if ( $result->{'status'} eq 'OK' );
+            print "\n";
+            for my $rr ( @{ $result->{'names'} } ) {
+                printf '%3s %-18s <%02s> %-18s', '', $rr->{'name'},
+                    $rr->{'suffix'},
+                    $rr->{'G'};
+                print "\n";
+            }
+            $cv->send;
+        }
+    );
+
+    $cv->recv;
+
+=head1 DESCRIPTION
+
+L<App::Netdisco::AnyEvent::Nbtstat> is an asynchronous AnyEvent NetBIOS node
+status requester.
+
+=head1 ATTRIBUTES
+
+L<App::Netdisco::AnyEvent::Nbtstat> implements the following attributes.
+
+=head2 C<interval>
+
+    my $interval = $request->interval;
+    $request->interval(1);
+
+Interval between requests, defaults to 0.02 seconds.
+
+=head2 C<timeout>
+
+    my $timeout = $request->timeout;
+    $request->timeout(2);
+
+Maximum request response time, defaults to 0.5 seconds.
+
+=head1 METHODS
+
+L<App::Netdisco::AnyEvent::Nbtstat> implements the following methods.
+
+=head2 C<nbtstat>
+
+    $request->nbtstat($ip, sub {
+        my $result = shift;
+    });
+
+Perform a NetBIOS node status request of $ip.
+
+=head1 SEE ALSO
+
+L<AnyEvent>
+
+=cut

Netdisco/lib/App/Netdisco/Core/Nbtstat.pm

@@ -5,11 +5,11 @@ use Dancer::Plugin::DBIC 'schema';
 
 use App::Netdisco::Util::Node 'check_mac';
 use NetAddr::IP::Lite ':lower';
-use Net::NBName;
+use App::Netdisco::AnyEvent::Nbtstat;
 
 use base 'Exporter';
 our @EXPORT = ();
-our @EXPORT_OK = qw/ do_nbtstat store_nbt /;
+our @EXPORT_OK = qw/ nbtstat_resolve_async store_nbt /;
 our %EXPORT_TAGS = (all => \@EXPORT_OK);
 
 =head1 NAME
@@ -25,42 +25,64 @@ subroutines.
 
 =head1 EXPORT_OK
 
-=head2 do_nbtstat( $node ) 
+=head2 nbtstat_resolve_async( $ips )
 
-Connects to node and gets NetBIOS information. Then adds entries to
+This method uses an asynchronous AnyEvent NetBIOS node status requester
-node_nbt table.
+C<App::Netdisco::AnyEvent::Nbtstat>.
 
-Returns whether a node is answering netbios calls or not.
+Given a reference to an array of hashes will connects to the C<IPv4> of a
+node and gets NetBIOS node status information.
+
+Returns the supplied reference to an array of hashes with MAC address,
+NetBIOS name, NetBIOS domain/workgroup, NetBIOS user, and NetBIOS server
+service status for addresses which responded.
 
 =cut
 
-sub do_nbtstat {
+sub nbtstat_resolve_async {
-    my ($host, $now) = @_;
+    my $ips = shift;
-    my $ip = NetAddr::IP::Lite->new($host) or return;
 
-    unless ( $ip->version() == 4 ) {
+    my $timeout  = setting('nbtstat_timeout')  || 1;
-        debug ' nbtstat only supports IPv4, invalid ip %s', $ip->addr;
+    my $interval = setting('nbtstat_interval') || 0.02;
-        return;
+
+    my $stater = App::Netdisco::AnyEvent::Nbtstat->new(
+        timeout  => $timeout,
+        interval => $interval
+    );
+
+    # Set up the condvar
+    my $cv = AE::cv;
+    $cv->begin( sub { shift->send } );
+
+    foreach my $hash_ref (@$ips) {
+        my $ip = $hash_ref->{'ip'};
+        $cv->begin;
+        $stater->nbtstat(
+            $ip,
+            sub {
+                my $res = shift;
+                _filter_nbname( $ip, $hash_ref, $res );
+                $cv->end;
+            }
+        );
     }
 
-    my $nb = Net::NBName->new;
+    # Decrement the cv counter to cancel out the send declaration
-    my $ns = $nb->node_status( $ip->addr );
+    $cv->end;
 
-    # Check for NetBIOS Info
+    # Wait for the resolver to perform all resolutions
-    return unless $ns;
+    $cv->recv;
 
-    my $nbname = _filter_nbname( $ip->addr, $ns );
+    # Close sockets
+    undef $stater;
 
-    if ($nbname) {
+    return $ips;
-        store_nbt($nbname, $now);
-    }
-
-    return 1;
 }
 
 # filter nbt names / information
 sub _filter_nbname {
     my $ip          = shift;
+    my $hash_ref = shift;
     my $node_status = shift;
 
     my $server = 0;
@@ -68,10 +90,10 @@ sub _filter_nbname {
     my $domain = '';
     my $nbuser = '';
 
-    for my $rr ( $node_status->names ) {
+    for my $rr ( @{$node_status->{'names'}} ) {
-        my $suffix = defined $rr->suffix ? $rr->suffix : -1;
+        my $suffix = defined $rr->{'suffix'} ? $rr->{'suffix'} : -1;
-        my $G      = defined $rr->G      ? $rr->G      : '';
+        my $G      = defined $rr->{'G'}      ? $rr->{'G'}      : '';
-        my $name   = defined $rr->name   ? $rr->name   : '';
+        my $name   = defined $rr->{'name'}   ? $rr->{'name'}   : '';
 
         if ( $suffix == 0 and $G eq "GROUP" ) {
             $domain = $name;
@@ -88,11 +110,11 @@ sub _filter_nbname {
     }
 
     unless ($nbname) {
-        debug ' nbtstat no computer name found for %s', $ip;
+      debug sprintf ' nbtstat no computer name found for %s', $ip;
         return;
     }
 
-    my $mac = $node_status->mac_address || '';
+    my $mac = $node_status->{'mac_address'} || '';
 
     unless ( check_mac( $ip, $mac ) ) {
 
@@ -101,23 +123,23 @@ sub _filter_nbname {
             ->single( { ip => $ip, -bool => 'active' } );
 
         if ( !defined $node_ip ) {
-            debug ' no MAC for %s returned by nbtstat or in DB', $ip;
+            debug sprintf ' no MAC for %s returned by nbtstat or in DB', $ip;
             return;
         }
         $mac = $node_ip->mac;
     }
 
-    return {
+        $hash_ref->{'ip'} = $ip;
-        ip     => $ip,
+        $hash_ref->{'mac'}    = $mac;
-        mac    => $mac,
+        $hash_ref->{'nbname'} = $nbname;
-        nbname => $nbname,
+        $hash_ref->{'domain'} = $domain;
-        domain => $domain,
+        $hash_ref->{'server'} = $server;
-        server => $server,
+        $hash_ref->{'nbuser'} = $nbuser;
-        nbuser => $nbuser
+        
-    };
+    return;
 }
 
-=head2 store_nbt($nb_hash_ref, $now?)
+=item store_nbt($nb_hash_ref, $now?)
 
 Stores entries in C<node_nbt> table from the provided hash reference; MAC
 C<mac>, IP C<ip>, Unique NetBIOS Node Name C<nbname>, NetBIOS Domain or

Netdisco/lib/App/Netdisco/Daemon/Worker/Poller/Nbtstat.pm

@@ -3,7 +3,7 @@ package App::Netdisco::Daemon::Worker::Poller::Nbtstat;
 use Dancer qw/:moose :syntax :script/;
 use Dancer::Plugin::DBIC 'schema';
 
-use App::Netdisco::Core::Nbtstat 'do_nbtstat';
+use App::Netdisco::Core::Nbtstat qw/nbtstat_resolve_async store_nbt/;
 use App::Netdisco::Util::Node 'is_nbtstatable';
 use App::Netdisco::Util::Device qw/get_device is_discoverable/;
 use App::Netdisco::Daemon::Util ':all';
@@ -33,7 +33,7 @@ sub nbtstat  {
   }
 
   # get list of nodes on device
-  my $interval = (setting('nbt_max_age') || 7) . ' day';
+  my $interval = (setting('nbtstat_max_age') || 7) . ' day';
   my $rs = schema('netdisco')->resultset('NodeIp')->search({
     -bool => 'me.active',
     -bool => 'nodes.active',
@@ -46,10 +46,25 @@ sub nbtstat  {
   })->ip_version(4);
 
   my @nodes = $rs->get_column('ip')->all;
-  my $now = 'to_timestamp('. (join '.', gettimeofday) .')';
 
-  $self->_single_node_body('nbtstat', $_, $now)
+  # Unless we have IP's don't bother
-    for @nodes;
+  if (scalar @nodes) {
+    # filter exclusions from config
+    @nodes = grep { is_nbtstatable( $_ ) } @nodes;
+
+    # setup the hash nbtstat_resolve_async expects
+    my @ips = map {+{'ip' => $_}} @nodes;
+    my $now = 'to_timestamp('. (join '.', gettimeofday) .')';
+
+    my $resolved_nodes = nbtstat_resolve_async(\@ips);
+
+    # update node_nbt with status entries
+    foreach my $result (@$resolved_nodes) {
+      if (defined $result->{'nbname'}) {
+        store_nbt($result, $now);
+      }
+    }
+  }
 
   return job_done("Ended nbtstat for ". $host->addr);
 }

Netdisco/lib/App/Netdisco/Manual/Configuration.pod

@@ -681,6 +681,20 @@ Value: Number. Default: 7.
 The maximum age of a node in days for it to be checked for NetBIOS
 information.
 
+=head3 C<nbtstat_interval>
+
+Value: Number. Default: 0.02.
+
+Interval between nbtstat requests in each poller. Defaults to 0.02 seconds,
+equating to 50 requests per second per poller.
+
+=head3 C<nbtstat_timeout>
+
+Value: Number. Default: 1.
+
+Seconds nbtstat will wait for a response before time out.  Accepts fractional
+seconds as well as integers.
+
 =head3 C<expire_devices>
 
 Value: Number of Days.

Netdisco/lib/App/Netdisco/Util/Node.pm

@@ -221,11 +221,9 @@ Returns false if the host is not permitted to nbtstat the target node.
 sub is_nbtstatable {
   my $ip = shift;
 
-  return _bail_msg("is_nbtstatable: node matched nbtstat_no")
+  return if check_node_no($ip, 'nbtstat_no');
-    if check_node_no($ip, 'nbtstat_no');
 
-  return _bail_msg("is_nbtstatable: node failed to match nbtstat_only")
+  return unless check_node_only($ip, 'nbtstat_only');
-    unless check_node_only($ip, 'nbtstat_only');
 
   return 1;
 }

Netdisco/share/config.yml

@@ -119,6 +119,8 @@ arpnip_min_age: 0
 nbtstat_no: []
 nbtstat_only: []
 nbtstat_max_age: 7
+nbtstat_interval: 0.02
+nbtstat_timeout: 1
 expire_devices: 0
 expire_nodes: 0
 expire_nodes_archive: 0