Net::Connection::Sniffer::Report - network profiling reports

  use Net::Connection::Sniffer::Report qw(
	read_stf
	by_cidr24
	by_name
	report
	presync
	sync
	chkcache
	get_lock
	chk_wconf
	rem_wchk
	rqst_dump
	web_report
	my_time
	dyn_bind
	xhandle
	rem_dump
	rem_update
	rem_report
  };

  ($stats,$head) = read_stf($filepathname);
  ($byc24,$rate,$bw) = by_cidr24($stats);
  ($byname,$rate,$bw) = by_name($stats,$byc24);
   $rv = report(*HANDLE,$file,$type);
   $rv = presync($filepathname);
   $rv = sync($filepathname,$oldtime,$timeout);
   $rv = chkcache($filepathname,$age);
  ($lock,$file) = get_lock($filepathname,$timeout,$umask);
   $rv = chk_wconf($conf);
   $rv = rem_wchk($conf);
   $rv = rqst_dump($sin,$file,$age,$updto);
   web_report($wconf);
   $timestring = my_time($epoch_seconds);
   $port = dyn_bind($sock,$iaddr);
   $handle = xhandle($program_string);
   $rv = rem_dump($conf);
   $rv = rem_update($config);
   rem_report($wconf);

	read_stf
	by_cidr24
	by_name
	report
	presync
	sync
	chkcache
	get_lock
	chk_wconf
	rem_wchk
	rqst_dump
	web_report
	my_time
	dyn_bind
	xhandle
	rem_dump
	rem_update
	rem_report

#!/usr/bin/perl
package Net::Connection::Sniffer::Report;

use strict;
#use diagnostics;

use Net::NBsocket qw(
	open_udpNB
	sockaddr_in
	inet_ntoa
	inet_aton
);
use Sys::Hostname::FQDN qw(
	fqdn
);

# pack sockaddr_in
# 0     pfinet  2
# 1             0
# 2     port    0
# 3             1
# 4     naddr   0
# 5             1
# 6             2
# 7             3
# 8     pad     0
# 9     to      0
# 10    16      0
# 11    bytes   0
# 12            0
# 13            0
# 14            0
# 15            0
#      $sin = pack('C2na4L2',$pf_inet,0,$2,inet_aton($1),0,0)

use NetAddr::IP::Util qw(
	inet_any2n
	sub128
	hasbits
);
use Net::Connection::Sniffer::Util;

use vars qw($VERSION @ISA @EXPORT_OK);

$VERSION = do { my @r = (q$Revision: 0.09 $ =~ /\d+/g); sprintf "%d."."%02d" x $#r, @r };

require Exporter;
@ISA = qw(Exporter);

@EXPORT_OK = qw(
	read_stf
	by_cidr24
	by_name
	report
	presync
	sync
	chkcache
	get_lock
	chk_wconf
	rem_wchk
	rqst_dump
	web_report
	my_time
	dyn_bind
	xhandle
	rem_dump
	rem_update
	rem_report
);

sub read_stf {
  local *F;
  my $file = shift;
  eval {
    die "no status file '$file'"
	unless $file && -e $file && -r $file && -s $file &&
	open (F,$file);
  };
  if ($@) {
    return wantarray ? () : undef;
  }

  my $header = '';
  while (($_ = <F>) =~ /^#/ || $_ !~ /\S/) {
    $header .= $_;
  }
  local $/ = undef;
  $_ .= <F>;
  close F;

  return eval "$_"
	unless wantarray;
  return (eval "$_", $header);
}

sub by_cidr24 {
  my $s = shift;
  my $bandwidth = 0;
  my $rate = 0;
  my($ln,$r,$w,$e,$nip);
  my($ce,$cr,$cw,$cn,$key);
  my $num = 0;
  my $composite = {};
  foreach $_ (sort keys %$s) {
    $nip = newcidr24 Net::Connection::Sniffer::Util(inet_any2n($_));
    $r = $s->{$_}->{R};
    $w = $s->{$_}->{W};
    $e = $s->{$_}->{E};
    unless ($ln) {		# first time through
      $cr = $r;
      $cw = $w;
      $ce = $e;
      $ln = $nip;
      $key = [$_];
    }
    elsif ($nip->within($ln)) {
      $cr += $r;
      $cw += $w;
      $ce = $e if $e > $ce;
      push @{$key}, $_;
    } else {
      $ln = $nip;
      $composite->{$num}->{R} = $cr;
      $composite->{$num}->{W} = $cw;
      $composite->{$num}->{E} = $ce;
      @_ = sort {		# put ip's in numeric order
	inet_aton($a) cmp inet_aton($b)
      } @$key;
      $composite->{$num}->{A} = [@_];
      $rate += $cr;
      $bandwidth += $cw;
      $num++;
      $cr = $r;
      $cw = $w;
      $ce = $e;
      $key = [$_];
    }
  }
  if ($nip->equal($ln)) {
    $composite->{$num}->{R} = $cr;
    $composite->{$num}->{W} = $cw;
    $composite->{$num}->{E} = $ce;
    @_ = sort {		# put ip's in numeric order
      inet_aton($a) cmp inet_aton($b)
    } @$key;
    $composite->{$num}->{A} = [@_];
    $rate += $cr;
    $bandwidth += $cw;
  }
  return ($composite,$rate,$bandwidth);
}

sub by_name {
  my($s,$bc) = @_;
  my $byname = {};
  my $lastdom = '';
  my $rate = 0;
  my $bw = 0;
  foreach(sort {
	my $A = $bc->{$a}->{A}->[0];	# use the first IP address to get a host name
	my $B = $bc->{$b}->{A}->[0];
	(lc $s->{$A}->{N}->[0] || '') cmp (lc $s->{$B}->{N}->[0] || '')
    } keys %$bc) {
    my $ip = $bc->{$_}->{A}->[0];
    my $name = reverse split('',lc $s->{$ip}->{N}->[0]);
    my $ndm;
    if ($name =~ /^[\w-]{2}\.[\w-]{3}\.[\w-]+/) {	# of the form 'name.(net|com|org).countrycode'
      $ndm = $&;
    }
    elsif ( $name =~ /^[\w-]+\.[\w-]+/) {		# of the form 'name.ext'
      $ndm = $&;
    } else {
      $ndm = $name;					# don't know what it is, leave it
    }
    if ($lastdom eq $ndm) {
      if ($bc->{$_}->{E} > $byname->{$ndm}->{E}) {
	$byname->{$ndm}->{E} = $bc->{$_}->{E};
      }
      $byname->{$ndm}->{R} += $bc->{$_}->{R};
      $byname->{$ndm}->{W} += $bc->{$_}->{W};
      push @{$byname->{$ndm}->{A}},@{$bc->{$_}->{A}};
    } else {	# new entry
      $lastdom = $ndm;					# new, update
      $byname->{$ndm}->{E} = $bc->{$_}->{E};
      $byname->{$ndm}->{R} = $bc->{$_}->{R};
      $byname->{$ndm}->{W} = $bc->{$_}->{W};
      @{$byname->{$ndm}->{A}} = @{$bc->{$_}->{A}};
    }
    $rate += $bc->{$_}->{R};
    $bw += $bc->{$_}->{W};
  }
  return ($byname,$rate,$bw);
}

sub report {
  my($handle,$file,$type) = @_;
  my($stats,$head) = read_stf($file);
  return $@ if $@;
  my($composite,$rate,$bandwidth) = by_cidr24($stats);

  if ($type) {
    ($composite,$rate,$bandwidth) = by_name($stats,$composite);
    $type = 'SUB DOMAIN';
  }
  else {
    $type = 'CIDR/24';
  }
  print $handle $head,
		 "\nComposite report, $type ordered by BANDWIDTH\n";
  my $title	="\n   IP addr\t    host name".
		 "\n qrys/mn -- bytes/sec -- #IP's\tlast contact time\n\n";
  print $handle $title;

  foreach my $key (sort {
	     (int ($composite->{$b}->{W}/3600) <=> int($composite->{$a}->{W}/3600))
						||
	 inet_aton($composite->{$a}->{A}->[0]) cmp inet_aton($composite->{$b}->{A}->[0])
	} keys %$composite) {
    foreach (sort {		# sort into ascending ip address order
	inet_aton($a) cmp inet_aton($b)
      } @{$composite->{$key}->{A}}) {
      my $name = join("\n\t\t",@{$stats->{$_}->{N}});
      printf $handle ("%s\t%s\n",$_,$name);
    }
    printf $handle (" %6.0f\t -- %.0f\t -- %d\t%s\n",$composite->{$key}->{R} /60,$composite->{$key}->{W} /3600,scalar @{$composite->{$key}->{A}},my_time($composite->{$key}->{E}));
  }
  print $handle $title;
  printf $handle ("rate\t\t= %.0f connections per minute\nbandwidth\t= %.0f bytes per second\nusers\t\t= %d\n",$rate /60,$bandwidth /3600,scalar keys %$stats);
  return undef;
}

# return at beginning of next second
#
sub _next_sec {
  my ($then) = @_;
  $then = time unless $then;
  my $now;
# wait for epoch
  do { select(undef,undef,undef,0.1); $now = time }
        while ( $then >= $now );
  $now;
}

# return file ctime
#
sub _ctime {
  return (stat($_[0]))[10];
}

sub presync {
  my ($file) = @_;
  return undef unless -e $file && -r $file;
  my $now = time;
  my $ctime = _ctime($file);
  return $ctime if $ctime < $now;
  $now = _next_sec($now);
  return undef if $ctime > $now;	# error if someone messes with ctime
  $ctime;
}

sub sync {
  my($file,$old,$to) = @_;
  $old = 0 unless $old;
  $to = 30 unless $to;
  $file =~ m|[^/]+$|;
  return undef unless -d $` && -r $`;		# directory must exist and be readable
  my $ctime = (-e $file)
	? _ctime($file)
	: 0;
  return $ctime if $old && $ctime > $old;
  eval {
	local $SIG{ALRM} = sub {die "'$file' timeout"};
	alarm $to;
	while (1) {
	  last if $ctime && $ctime > $old;
	  select(undef,undef,undef,0.1);
	  $ctime = _ctime($file);
	}
	alarm 0;
  };
  return undef if $@;
  return $ctime;
}

sub chkcache {
  my($file,$age) = @_;
  return undef unless -e $file && -r $file;
  $age = 300 unless $age;
  my $ctime = _ctime($file);
  return undef if ($ctime + $age) < time;
  $ctime;
}

sub get_lock {
  my($file,$to,$umask) = @_;
  $to = 15 unless $to;
  $umask = 0117 unless $umask;
  require Fcntl;
  import Fcntl qw(:DEFAULT :flock);
  local(*LOCK,*FILE);
  my $lock;
  eval {
      local $SIG{ALRM} = sub {die "can not lock $file"};
      alarm $to;
      umask $umask;
      while (1) {
	die "could not open lock file '${file}.lock'"
		unless sysopen LOCK, $file .'.lock', &O_RDWR|&O_CREAT|&O_TRUNC;
	unless (flock LOCK, &LOCK_EX) {
	  close LOCK;
	  select(undef,undef,undef,1);		# sleep a second, don't use 'sleep'
	  next;
	}
# lock achieved
	unless (sysopen FILE, $file, &O_RDWR|&O_CREAT) {
	  close LOCK;
	  die "could not open '$file' after lock"
	}
	alarm 0;
	last;
      }
  };
  return $@ ? () : (*LOCK,*FILE);
}

sub chk_wconf {
  my $wc = shift;
  return 'required statistics file specification missing'
	unless exists $wc->{stats};
  $wc->{stats} =~ m|[^/]+$|;
  my $dir = $` || './';
  return "stats directory missing or not readable: '$dir'"
	unless -d $dir && -r $dir;
  if (exists $wc->{cache}) {
    $wc->{cache} =~ m|[^/]+$|;
    $dir = $` || './';
    return "cache directory not writable: '$dir'"
	unless -d $dir && -w $dir;
    if (-e $wc->{cache}) {
      return "invalid permissions on '$wc->{cache}': not writable"
	unless -w $wc->{cache};
    }
  }
  if (exists $wc->{updto} && $wc->{updto}) {
    return "invalid characters in update timeout: '$wc->{updto}'"
	if $wc->{updto} =~ /\D/;
  }
  else {
    $wc->{updto} = 15;		# default
  }
  if (exists $wc->{refresh} && $wc->{refresh}) {
    return "invalid characters in refresh: '$wc->{refresh}'"
	if $wc->{refresh} =~ /\D/;
  }
  else {
    $wc->{refresh} = 300;		# default
  }
  if (exists $wc->{update}) {
    if ($wc->{update} =~ /([\w.-]+)\:(\d+)/) {	# host + port	
      return "invalid characters in host address: '$1'"
	unless (my $iaddr = inet_aton($1));
      $wc->{update} = sockaddr_in($2,$iaddr);
    }
    elsif ($wc->{update} =~ /^\d+$/) {		# only digits
      $wc->{update} = sockaddr_in($&,&Socket::INADDR_LOOPBACK);
    }
    else {
      return "invalid update specification '$wc->{update}'";
    }
  }	# else it is disabled, must use kill -USR1
  return '';
}

sub rem_wchk {
  my ($c) = @_;
  return 'missing cache file entry'
	unless $c->{cache};
  $c->{stats} = $c->{cache}		# fudge stats entry for conf check if no localhost entry
	unless exists $c->{update} && $c->{update};
  goto &chk_wconf;
}

sub rqst_dump {
  my($theirsin,$file,$age,$updto) = @_;
# pre-sync the stats file
  my $ctime = presync($file) || 0;	# ignore errors if file is not yet present
  my $tmp = '';
  if  ($theirsin && (! $age || ($ctime + $age) < time)) {
    my $sock = open_udpNB();
    unless ($sock) {
      $tmp = 'could not open UDP socket';
      last;
    }
    while (1) {
      my $name = fqdn();
      unless (dyn_bind($sock,inet_aton($name))) {
	$tmp = "could not bind socket to $name, $!";
	last;
      }
      $tmp = 'dump';
      send($sock,$tmp,0,$theirsin);	# udp may not block
# wait for stats update
      $ctime = sync($file,$ctime,$updto);
      if ($@) {
	$tmp = $@;					# probably a timeout error
	last;
      }
      if (defined recv($sock,$tmp,256,0) &&		# retrieve response if we can
	$tmp =~ /OK\s*(\d+)/ &&			# good receive response
	$1 > $ctime) {				# weird timing error
	$tmp = "statistics update error OK $1 > file $ctime";
      } else {
	$tmp = '';
      }
      last;
    }
    close $sock;
  }
  return $tmp;
}

sub web_report {
  my $wconf = shift;
  my $type = $ARGV[0] ? 1 : 0;
  my $txt = 'failed: nothing processed';
  while (1) {
    last if ($txt = chk_wconf($wconf));	# exit on bad config file
    my $sin = $wconf->{update} || 0;
    if ($wconf->{cache}) {		# if cache exists, it has been checked
      my $cfile = $wconf->{cache} .'.'. $type;
      if (chkcache($cfile,$wconf->{refresh})) {
	eval {
		local *CACHE;
		local $/ = undef;
		open(CACHE,$cfile) or die "could not open cache '$cfile'";
		$txt = <CACHE>;
		close CACHE;
	};
        $txt = $@ if $@;
       last;
      }
      elsif ($sin) {	# dump enabled
	$txt = 'unknown dump request response';
# sync stats file and unconditionally request a dump
	last if ($txt = rqst_dump($sin,$wconf->{stats},0,$wconf->{updto}));
# lock cache file
	my($lock,$file) = get_lock($cfile .'.tmp',$wconf->{updto},0117);
	if ($@) {						# punt if lock error
	  $txt = $@;
	  last;
	}
	$txt = report($file,$wconf->{stats},$type);	# write report to cache
	close $file;
	if ($txt) {					# there was an error
	  close $lock;
	  last;
	}
	rename $cfile .'.tmp', $cfile;			# atomic move
	close $lock;
      }
      else {
	;	# DUMP is not enabled, do we want an error code? maybe they used SIG USR1
      }
      local *REPORT;
      if (open(REPORT,$cfile)) {
	foreach(<REPORT>) {
	  print $_;					# print to STDOUT
	}
	close REPORT;
      } else {
	$txt = "could not open '$cfile'";
      }
      last;
    }
    elsif ($sin) {	# cache file not configured, generate report directly from stats file
      last if ($txt = rqst_dump($sin,$wconf->{stats},$wconf->{refresh},$wconf->{updto}));
    } else {
      ;		# DUMP is not enabled, do we want an error code? maybe they used SIG USR1
    }
    $txt = report(*STDOUT,$wconf->{stats},$type);	# report to stdout
    last;
  } # end while(1)
  print $txt,"\n" if $txt;	# spit out any errors
}

my @months = qw(jan feb mar apr may jun jul aug sep oct nov dec);

sub my_time {
  my($time) = shift;
  my ($sec,$min,$hour,$mday,$mon) = localtime($time);
  return sprintf("%s %02d %02d:%02d:%02d",$months[$mon],$mday,$hour,$min,$sec);
}
 

sub dyn_bind {	# t => s_make_kid_Dbind.t
  my($sock,$iaddr) = @_;
  foreach(1..100) {
    my $port = 49152 + int rand(65536 - 49152);
    return $port if bind($sock,sockaddr_in($port,$iaddr));
  }
  return undef;
}

sub xhandle {
  local *xHandle;
  if (open(xHandle,$_[0] .'|')) {
    return *xHandle;
  } else {
    return undef;
  }
}

# subroutine to dump stats files on remote (and local) hosts
#
# input:	config hash
# returns:	true on success
#
sub rem_dump {
  my $wconf = shift;
  my $ssh = $wconf->{ssh} .' ';
  my $rsync = $wconf->{rsync} .' -utz -e '. $ssh .' ';
  $wconf->{cache} =~ m|[^/]+$|;
  my $dir = $` || './';
  my %src;
  local(*SDTERR);		# redirect error messages from rsync and ssh
  open STDERR, '>/dev/null';
  foreach (keys %{$wconf->{src}}) {
    $src{$_} = xhandle($ssh . $_ .' '. $wconf->{src}->{$_}->{exec});
  }
  if (exists $wconf->{update}) {
    rqst_dump($wconf->{update},$wconf->{stats},$wconf->{refresh},$wconf->{updto});
  }
  my $debugtxt = '';
  foreach (keys %src) {		# wait for ssh completion
    my $xhndl = $src{$_};
    next unless $xhndl;
    undef local $/;
    $debugtxt .= <$xhndl>;
    close $xhndl;
  }
  foreach (keys %src) {
    $src{$_} = xhandle($rsync . $_ .':'. $wconf->{src}{$_}->{fetch} .' '. $dir .'/'. $_ . '.stats');
  }
  foreach (keys %src) {		# wait for rsync completion
    my $xhndl = $src{$_};
    next unless $xhndl;    
    undef local $/;
    $debugtxt .= <$xhndl>;    
    close $xhndl;    
  }
  return 1;
}

sub rem_update {
  require Data::Dumper;
  my $c = shift;
  my $cache = $c->{cache};
  return undef unless $cache =~ m|/[\w\.\-]+$|;
  $c->{cache} =~ m|[^/]+$|;
  my $dir = $` || './';
  $cache .= '.1';		# unconditionally type1
  my $comp = $dir .'composite.stats';
  my @files;
  foreach (keys %{$c->{src}}) {
    push @files, $dir .'/'. $_ .'.stats';
  }
  if (exists $c->{update}) {	# update local host if configured
    push @files, $c->{stats};
  }
#               B => 5678,	    # bytes accumulated
#               C => 1234,	    # counts
#               E => 124444,	    # count epoch
#               N => ['hostname',], # hostname(s) for this IP
#               R => 2345,	    # rate
#               S => 123456,	    # count start time
#               T => 123455,	    # TTL timeout of PTR record
#               W => 7890,	    # bandWidth

  my $hits = 0;
  my $bytes = 0;
  my $users = 0;
  my $txt = '';
  my $stats = {};
  foreach (@files) {
    my($sf,$ft) = read_stf($_);
    $hits += $1 if $ft =~ /hits:\s+(\d+)/;
    $bytes += $1 if $ft =~ /bytes:\s+(\d+)/;
# some users may be in all hosts, this is not a reasonable sum
#    $users += $1 if $ft =~ /users:\s+(\d+)/;
    $txt .= $ft;
    foreach (keys %$sf) {
      if (exists $stats->{$_}) {
        $stats->{$_}->{B} += $sf->{$_}->{B};
        $stats->{$_}->{C} += $sf->{$_}->{C};
        $stats->{$_}->{E} = $sf->{$_}->{C}
		if $stats->{$_}->{E} < $sf->{$_}->{C};
        $stats->{$_}->{R} += $sf->{$_}->{R};
        $stats->{$_}->{W} += $sf->{$_}->{W};
	my $ipk = $_;
	foreach my $n (@{$sf->{$ipk}->{N}}) {
	  unless (grep(/$n/,@{$stats->{$ipk}->{N}})) {
	    push @{$stats->{$ipk}->{N}}, $n;
	  }
	}
      }
      else {
        $stats->{$_} = $sf->{$_};
      }
    }
  }

  local *CACHE;
  open (CACHE,'>'. $comp .'.tmp');
  print CACHE $txt;
  print CACHE qq|
# grand total all hosts
# hits: \t$hits per minute
# bytes:\t$bytes per second
|;
  print CACHE 'my ',&Data::Dumper::Dumper($stats);
  close CACHE;
  rename $comp .'.tmp', $comp;

  open (CACHE,'>'. $cache .'.tmp');
  print report(*CACHE,$comp,1),"\n";
  close CACHE;
  rename $cache .'.tmp', $cache;
}

sub rem_report {
  my $wconf = shift;
  my $txt = '';
  my $type = 1;			# unconditional for remote reports
  while (1) {
    last if ($txt = rem_wchk($wconf));	# exit on bad config file
    my $cfile = $wconf->{cache} .'.'. $type;
    if (chkcache($cfile,$wconf->{refresh})) {
	eval {
		local *CACHE;
		local $/ = undef;
		open(CACHE,$cfile) or die "could not open cache '$cfile'";
		$txt = <CACHE>;
		close CACHE;
	};
        $txt = $@ if $@;
	last;
    }
    unless (rem_dump($wconf)) {
      $txt = 'could not dump remotes';
      last;
    }
    unless (rem_update($wconf)) {
      $txt = 'could not update composite stats';
      last;
    }
    local *REPORT;
    if (open(REPORT,$cfile)) {
      foreach(<REPORT>) {
        print $_;                                     # print to STDOUT
      }
      close REPORT;
    } else {
      $txt = "could not open '$cfile'";
    }
    last;
  } # end while(1)
  print $txt,"\n" if $txt;	# spit out any errors
}

1;