#  Cflow.pm - perl module for processing cflowd raw flow files
#  Copyright (C) 1998-2000  Dave Plonka
#
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 2 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.

# $Id: Cflow.pm,v 1.27 2001/02/08 17:22:18 dplonka Exp $
# Dave Plonka <plonka@doit.wisc.edu>

use strict;

# got these from "netinet/tcp.ph" (from "h2ph netinet/tcp.h")
$Cflow::TH_FIN = 0x01;
$Cflow::TH_SYN = 0x02;
$Cflow::TH_RST = 0x04;
$Cflow::TH_PUSH = 0x08;
$Cflow::TH_ACK = 0x10;
$Cflow::TH_URG = 0x20;

my %bits = (
   FIN => $Cflow::TH_FIN,
   SYN => $Cflow::TH_SYN,
   RST => $Cflow::TH_RST,
   PUSH => $Cflow::TH_PUSH,
   ACK => $Cflow::TH_ACK,
   URG => $Cflow::TH_URH
);

# got these from "netinet/ip_icmp.ph" (from "h2ph netinet/ip_icmp.h")

# { ICMP Types:
$Cflow::ICMP_ECHOREPLY = 0;
$Cflow::ICMP_DEST_UNREACH = 3;
$Cflow::ICMP_SOURCE_QUENCH = 4;
$Cflow::ICMP_REDIRECT = 5;
$Cflow::ICMP_ECHO = 8;
$Cflow::ICMP_TIME_EXCEEDED = 11;
$Cflow::ICMP_PARAMETERPROB = 12;
$Cflow::ICMP_TIMESTAMP = 13;
$Cflow::ICMP_TIMESTAMPREPLY = 14;
$Cflow::ICMP_INFO_REQUEST = 15;
$Cflow::ICMP_INFO_REPLY = 16;
$Cflow::ICMP_ADDRESS = 17;
$Cflow::ICMP_ADDRESSREPLY = 18;
# }{ Codes for UNREACH:
$Cflow::ICMP_NET_UNREACH = 0;
$Cflow::ICMP_HOST_UNREACH = 1;
$Cflow::ICMP_PROT_UNREACH = 2;
$Cflow::ICMP_PORT_UNREACH = 3;
$Cflow::ICMP_FRAG_NEEDED = 4;
$Cflow::ICMP_SR_FAILED = 5;
$Cflow::ICMP_NET_UNKNOWN = 6;
$Cflow::ICMP_HOST_UNKNOWN = 7;
$Cflow::ICMP_HOST_ISOLATED = 8;
$Cflow::ICMP_NET_ANO = 9;
$Cflow::ICMP_HOST_ANO = 10;
$Cflow::ICMP_NET_UNR_TOS = 11;
$Cflow::ICMP_HOST_UNR_TOS = 12;
$Cflow::ICMP_PKT_FILTERED = 13;
$Cflow::ICMP_PREC_VIOLATION = 14;
$Cflow::ICMP_PREC_CUTOFF = 15;
$Cflow::ICMP_UNREACH = 15;
# }{ Codes for REDIRECT:
$Cflow::ICMP_REDIR_NET = 0;
$Cflow::ICMP_REDIR_HOST = 1;
$Cflow::ICMP_REDIR_NETTOS = 2;
$Cflow::ICMP_REDIR_HOSTTOS = 3;
# }{ Codes for TIME_EXCEEDED:
$Cflow::ICMP_EXC_TTL = 0;
$Cflow::ICMP_EXC_FRAGTIME = 1;
# }

package Cflow::SymbolicTCPFlags; ###############################################
use Carp;

sub TIESCALAR {
   my $class = shift;
   die unless $class;
   my $this = shift;
   die unless ref($this);
   bless $this, $class
}

sub FETCH {
   my $this = shift;
   my($symbolic_tcp_flags) = '';
   my(@names);

   if (6 == $Cflow::protocol && 0x0 != $$this) {
      while (my($name, $value) = each(%bits)) {
	 push(@names, $name) if ($$this & $value)
      }
      $symbolic_tcp_flags = '(' . join('|', @names) . ')'
   }

   $symbolic_tcp_flags
}

sub STORE {
   croak "Can't modify read-only value in scalar assignment"
}

package Cflow::SymbolicICMPTypeCode; ###########################################
use Carp;

my @typecode = (
[ 'ECHOREPLY' ],    # 0 /* echo reply */
undef,              # 1
undef,              # 2
[ 'UNREACH',        # 3 /* dest unreachable, codes: */
   ['NET_',           # 0 /* bad net */
    'HOST_',          # 1 /* bad host */
    'PROTOCOL_',      # 2 /* bad protocol */
    'PORT_',          # 3 /* bad port */
    'NEEDFRAG_',      # 4 /* IP_DF caused drop */
    'SRCFAIL_',       # 5 /* src route failed */
    'NET_UNKNOWN_',   # 6
    'HOST_UNKNOWN_',  # 7
    'HOST_ISOLATED_', # 8
    'NET_ANO_',       # 9
    'HOST_ANO_',      # 10
    'NET_UNR_TOS_',   # 11
    'HOST_UNR_TOS_',  # 12
    'PKT_FILTERED_',  # 13 /* Packet filtered */
    'PREC_VIOLATION_', # 14 /* Precedence violation */
    'PREC_CUTOFF_',   # 15 /* Precedence cut off */
    ] ],
[ 'SOURCE_QUENCH' ], # 4 /* packet lost, slow down */
[ 'REDIRECT',       # 5 /* shorter route, codes: */
   ['NET_',           # 0 /* for network */
    'HOST_',          # 1 /* for host */
    'TOSNET_',        # 2 /* for tos and net */
    'TOSHOST_'] ],    # 3 /* for tos and host */
undef,              # 6
undef,              # 7
[ 'ECHO' ],         # 8 /* echo service */
undef,              # 9
undef,              # 10
[ 'TIME_EXCEEDED',  # 11 /* time exceeded, code: */
   ['INTRANS_',       # 0 /* ttl==0 in transit */
    'REASS_'] ],      # 1 /* ttl==0 in reass */
[ 'PARAMPROB' ],    # 12 /* ip header bad */
[ 'TIMESTAMP' ],    # 13 /* timestamp request */
[ 'TIMESTAMPREPLY' ], # 14 /* timestamp reply */
[ 'INFO_REQUEST' ], # 15 /* information request */
[ 'INFO_REPLY' ],   # 16 /* information reply */
[ 'MASKREQ' ],      # 17 /* address mask request */
[ 'MASKREPLY' ],    # 18 /* address mask reply */
);

sub TIESCALAR {
   my $class = shift;
   die unless $class;
   my $this = shift;
   die unless ref($this);
   bless $this, $class
}

sub FETCH {
   my $this = shift;
   return '' unless (1 == $Cflow::protocol);
   return $typecode[$Cflow::ICMPType]->[1]->[$Cflow::ICMPCode] . $typecode[$Cflow::ICMPType]->[0]
}

package Cflow::InetNtoA; #######################################################
use Carp;

sub TIESCALAR {
   my $class = shift;
   die unless $class;
   my $this = shift;
   die unless ref($this);
   bless $this, $class
}

sub FETCH {
   my $this = shift;
   join('.', unpack('C4', pack('N', $$this))) # inet_ntoa was too slow
}

sub STORE {
   croak "Can't modify read-only value in scalar assignment"
}

package Cflow::LocalTime; ######################################################
use Carp;
use POSIX; # for strftime

sub TIESCALAR {
   my $class = shift;
   die unless $class;
   my $this = shift;
   die unless ref($this);
   bless $this, $class
}

sub FETCH {
   my $this = shift;
   strftime("%Y/%m/%d %H:%M:%S", localtime($$this))
}

sub STORE {
   croak "Can't modify read-only value in scalar assignment"
}

package Cflow::PerSecond; ######################################################
use Carp;

sub TIESCALAR {
   my $class = shift;
   die unless $class;
   my $this = shift;
   die unless ref($this);
   bless $this, $class
}

sub FETCH {
   my $this = shift;
   my $seconds = $Cflow::endtime - $Cflow::startime;
   $$this / (($seconds > 0)? $seconds : 1)
}

sub STORE {
   croak "Can't modify read-only value in scalar assignment"
}

package Cflow::Octets2BitsPerSecond; ###########################################
use Carp;

sub TIESCALAR {
   my $class = shift;
   die unless $class;
   my $this = shift;
   die unless ref($this);
   bless $this, $class
}

sub FETCH {
   my $this = shift;
   my $seconds = $Cflow::endtime - $Cflow::startime;
   $$this / (($seconds > 0)? $seconds : 1)
}

sub STORE {
   croak "Can't modify read-only value in scalar assignment"
}

package Cflow; #################################################################

# convert the RCS revision to a reasonable Exporter VERSION:
'$Revision: 1.27 $' =~ m/(\d+)\.(\d+)/ && (( $Cflow::VERSION ) = sprintf("%d.%03d", $1, $2));

require Exporter;
@Cflow::ISA = qw(Exporter);
%Cflow::EXPORT_TAGS = (

flowvars => [qw(
$unix_secs
$exporter
$exporterip
$srcaddr
$srcip
$dstaddr
$dstip
$input_if
$output_if
$srcport
$dstport
$pkts
$bytes
$nexthop
$nexthopip
$startime
$endtime
$protocol
$tos $src_as
$dst_as
$src_mask
$dst_mask
$tcp_flags
$engine_type
$engine_id
$localtime
$raw
$Bps
$pps
$TCPFlags
$ICMPTypeCode
$ICMPType
$ICMPCode
)],

tcpflags => [qw(
$TH_FIN
$TH_SYN
$TH_RST
$TH_PUSH
$TH_ACK
$TH_URG
)],

icmptypes => [qw(
$ICMP_ECHOREPLY
$ICMP_DEST_UNREACH
$ICMP_SOURCE_QUENCH
$ICMP_REDIRECT
$ICMP_ECHO
$ICMP_TIME_EXCEEDED
$ICMP_PARAMETERPROB
$ICMP_TIMESTAMP
$ICMP_TIMESTAMPREPLY
$ICMP_INFO_REQUEST
$ICMP_INFO_REPLY
$ICMP_ADDRESS
$ICMP_ADDRESSREPLY
)],

icmpcodes => [qw(
$ICMP_NET_UNREACH
$ICMP_HOST_UNREACH
$ICMP_PROT_UNREACH
$ICMP_PORT_UNREACH
$ICMP_FRAG_NEEDED
$ICMP_SR_FAILED
$ICMP_NET_UNKNOWN
$ICMP_HOST_UNKNOWN
$ICMP_HOST_ISOLATED
$ICMP_NET_ANO
$ICMP_HOST_ANO
$ICMP_NET_UNR_TOS
$ICMP_HOST_UNR_TOS
$ICMP_PKT_FILTERED
$ICMP_PREC_VIOLATION
$ICMP_PREC_CUTOFF
$ICMP_UNREACH
$ICMP_REDIR_NET
$ICMP_REDIR_HOST
$ICMP_REDIR_NETTOS
$ICMP_REDIR_HOSTTOS
$ICMP_EXC_TTL
$ICMP_EXC_FRAGTIME
)]

);

@Cflow::EXPORT_OK = qw(find verbose);
# add the symbols for the specified tag(s) to @EXPORT_OK:
Exporter::export_ok_tags(qw(flowvars tcpflags icmptypes icmpcodes));

=head1 NAME

Cflow::find - find "interesting" flows in cflowd flow files

=head1 SYNOPSIS

   use Cflow;

   Cflow::verbose(1);
   Cflow::find(\&wanted, <*.flows*>);

   sub wanted { ... }

=head1 DESCRIPTION

Cflow::find() will iterate across all the flows in the specified files.
It will call your wanted() function once for each flow.  It returns
a string containing a ratio (similar to a hash value in a scalar context)
indicating ((# of "wanted" flows) / (# of scanned flows)).  The wanted()
function does whatever verifications you want.  Within your wanted()
function, tests on the "current" flow can be performed using the following
variables:

   $Cflow::unix_secs - secs since epoch
   $Cflow::exporter - Exporter IP Address as a host-ordered "long"
   $Cflow::exporterip - Exporter IP Address as dotted-decimal string
   $Cflow::localtime - $Cflow::unix_secs interpreted as localtime
		       with this strftime(3) format: "%Y/%m/%d %H:%M:%S"
   $Cflow::srcaddr - Source IP Address as a host-ordered "long"
   $Cflow::srcip - Source IP Address as a dotted-decimal string
   $Cflow::dstaddr - Destination IP Address as a host-ordered "long"
   $Cflow::dstip - Destination IP Address as a dotted-decimal string
   $Cflow::input_if - Input interface index
   $Cflow::output_if - Output interface index
   $Cflow::srcport - TCP/UDP src port number or equivalent
   $Cflow::dstport - TCP/UDP dst port number or equivalent
   $Cflow::ICMPType - high byte of $Cflow::dstport
		      Undefined if the current flow is not an ICMP flow.
   $Cflow::ICMPCode - low byte of $Cflow::dstport
		      Undefined if the current flow is not an ICMP flow.
   $Cflow::ICMPTypeCode - symbolic representation of $Cflow::dstport
			  The value is a the type-specific ICMP code,
			  if any, followed by the ICMP type.
			  (e.g. 'ECHO' or 'HOST_UNREACH')
			  Undefined if the current flow is not an ICMP
			  flow.
   $Cflow::pkts - Packets sent in Duration
   $Cflow::bytes - Octets sent in Duration
   $Cflow::nexthop - Next hop router's IP Address as a host-ordered "long"
   $Cflow::nexthopip - Next hop router's IP Address as a dotted-decimal
		       string
   $Cflow::startime - secs since epoch at start of flow
   $Cflow::endtime - secs since epoch at last packet of flow
   $Cflow::protocol - IP protocol, e.g., 6=TCP, 17=UDP, ...
   $Cflow::tos - IP Type-of-Service
   $Cflow::src_as - originating AS of source address
   $Cflow::dst_as - originating AS of destination address
   $Cflow::src_mask - source address prefix mask bits
   $Cflow::dst_mask - destination address prefix mask bits
   $Cflow::tcp_flags - bitwise OR of all TCP flags in flow;
		       0x10 for non-TCP flows
   $Cflow::TCPFlags - symbolic representation of $Cflow::tcp_flags
		      The value will be a bitwise-or expression.
		      (e.g. 'SYN|ACK')
		      Undefined if the current flow is not a TCP flow.
   $Cflow::engine_type - type of flow switching engine
   $Cflow::engine_id - ID of the flow switching engine
		       (Cisco version 5+ header)
   $Cflow::raw - the entire "packed" flow record as read from the
		 input file
		 This is useful when the "wanted" subroutine wants to
		 write the flow to another FILEHANDLE. i.e.:
		    syswrite(FILEHANDLE, $Cflow::raw, length $Cflow::raw)
   $Cflow::Bps - the minimum bytes per second for the current flow
   $Cflow::pps - the minimum packets per second for the current flow


If you'd like the find() function to return a useful "hit ratio" value,
your wanted() sub-routine should return a boolean value:  non-zero
indicating that the flow was indeed "wanted".  Then, find() will return
a "hit ratio" of a format similar to that of the value of a perl hash
when taken in a scalar context, e.g. "n/m".

Optionally, a reference to a perfile() function can be passed as the
argument following the reference to the wanted() function.  This
perfile() function will be called once for each flow file.  The
argument to the perfile() function will be name of th flow file
currently being read.  The primary purpose of this function is to
provide a mechanism by which Netflow::clearcache() can be called
periodically.  Otherwise it is probably of little use.

Since Cflow is an Exporter, you can request that all those scalar
flow variables be exported (so that you need not use the "Cflow::"
prefix):

   use Cflow qw(:flowvars);

Also, you can request that the symbolic names for the TCP flags,
ICMP types, and/or ICMP codes be exported:

   use Cflow qw(:tcpflags :icmptypes :icmpcodes);

The tcpflags are:

$TH_FIN $TH_SYN $TH_RST $TH_PUSH $TH_ACK $TH_URG

The icmptypes are:

$ICMP_ECHOREPLY     $ICMP_DEST_UNREACH $ICMP_SOURCE_QUENCH
$ICMP_REDIRECT      $ICMP_ECHO         $ICMP_TIME_EXCEEDED
$ICMP_PARAMETERPROB $ICMP_TIMESTAMP    $ICMP_TIMESTAMPREPLY
$ICMP_INFO_REQUEST  $ICMP_INFO_REPLY   $ICMP_ADDRESS
$ICMP_ADDRESSREPLY

The icmpcodes are:

$ICMP_NET_UNREACH  $ICMP_HOST_UNREACH $ICMP_PROT_UNREACH
$ICMP_PORT_UNREACH $ICMP_FRAG_NEEDED  $ICMP_SR_FAILED
$ICMP_NET_UNKNOWN  $ICMP_HOST_UNKNOWN $ICMP_HOST_ISOLATED
$ICMP_NET_ANO      $ICMP_HOST_ANO     $ICMP_NET_UNR_TOS
$ICMP_HOST_UNR_TOS $ICMP_PKT_FILTERED $ICMP_PREC_VIOLATION
$ICMP_PREC_CUTOFF  $ICMP_UNREACH      $ICMP_REDIR_NET
$ICMP_REDIR_HOST   $ICMP_REDIR_NETTOS $ICMP_REDIR_HOSTTOS
$ICMP_EXC_TTL      $ICMP_EXC_FRAGTIME

(Please note that the names above are not necessarily exactly the same
as the names of the flags, types, and codes as set in the values of the
aforemented $Cflow::TCPFlags and $Cflow::ICMPTypeCode flow variables.)

Lastly, as is usually the case for modules, the subroutine names can be
imported, and a minimum version of Cflow can be specified:

   use Cflow qw(:flowvars find verbose 1.023);

Here's a complete example with a sample wanted function.  It will print
information on all UDP flows that involve either a source or destination
port of 31337 and port on the other end that is unreserved (greater than
1024):

   use Cflow qw(:flowvars find);
   use POSIX; # for strftime

   my $udp = getprotobyname('udp');
   verbose(0);
   find(\&wanted, @ARGV? @ARGV : <*.flows*>);

   sub wanted {
      return if ($srcport < 1024 || $dstport < 1024);
      return unless ((($srcport == 31337 ||
		       $dstport == 31337) &&
		      $udp == $protocol));

      printf("%s %15.15s.%-5hu %15.15s.%-5hu %2hu %10u %10u\n",
	     $localtime,
	     $srcip,
	     $srcport,
	     $dstip,
	     $dstport,
	     $protocol,
	     $pkts,
	     $bytes)
   }

Cflow::verbose() takes a single scalar boolean argument and sets whether
or not you wish warning messages to be generated to STDERR when "errors"
occur.  Verbose mode is set by default. (Note: even under normal
circumstances, a cflowd flow file sometimes ends with partial data.  In
that instance, this package may generate erroneous warnings so take them
with a grain of salt or simply call Cflow::verbose(0).)

=head1 NOTES

Cflow::find() will generate warnings if the flow data file is "invalid" as far
as it's concerned. Currently it requires that you be using Cisco version 5
flow-export and specify that all flow-export data is saved using the COLLECT
field of the CISCOEXPORTER stanzas.  E.g.:

   COLLECT: { flows }

The interface presented by this package is a blatant ripoff of File::Find.

=head1 AUTHOR

Dave Plonka <plonka@doit.wisc.edu>

Copyright (C) 1998-2000  Dave Plonka.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.

=head1 VERSION

The version number is the module file RCS revision number (B<$Revision: 1.27 $>)
with the minor number printed right justified with leading zeroes to 3
decimal places.  For instance, RCS revision 1.1 would yield a package version
number of 1.001.

This is so that revision 1.10 (which is version 1.010), for example, will
test greater than revision 1.2 (which is version 1.002) when you want to
B<require> a minimum version of this module.

=cut

$Cflow::verbose = 1; # issue warnings to STDERR by default

# from "CflowdRawFlow.hh":
use constant k_routerMask        => 0x00000001;
use constant k_srcIpAddrMask     => 0x00000002;
use constant k_dstIpAddrMask     => 0x00000004;
use constant k_inputIfIndexMask  => 0x00000008;
use constant k_outputIfIndexMask => 0x00000010;
use constant k_srcPortMask       => 0x00000020;
use constant k_dstPortMask       => 0x00000040;
use constant k_pktsMask          => 0x00000080;
use constant k_bytesMask         => 0x00000100;
use constant k_ipNextHopMask     => 0x00000200;
use constant k_startTimeMask     => 0x00000400;
use constant k_endTimeMask       => 0x00000800;
use constant k_protocolMask      => 0x00001000;
use constant k_tosMask           => 0x00002000;
use constant k_srcAsMask         => 0x00004000;
use constant k_dstAsMask         => 0x00008000;
use constant k_srcMaskLenMask    => 0x00010000;
use constant k_dstMaskLenMask    => 0x00020000;
use constant k_tcpFlagsMask      => 0x00040000;
use constant k_inputEncapMask    => 0x00080000;
use constant k_outputEncapMask   => 0x00100000;
use constant k_peerNextHopMask   => 0x00200000;
use constant k_engineTypeMask    => 0x00400000;
use constant k_engineIdMask      => 0x00800000;

# This is the value for $Cflow::index that we *expect* using Cisco flow-export
# version 5 and cflowd-2.x.  This was determined by simple experimentation
# with just "COLLECT: { flows }" in "cflowd.conf".
$Cflow::expected_entry_mask =
   k_routerMask |
   k_srcIpAddrMask |
   k_dstIpAddrMask |
   k_inputIfIndexMask |
   k_outputIfIndexMask |
   k_srcPortMask |
   k_dstPortMask |
   k_pktsMask |
   k_bytesMask |
   k_ipNextHopMask |
   k_startTimeMask |
   k_endTimeMask |
   k_protocolMask |
   k_tosMask |
   k_srcAsMask |
   k_dstAsMask |
   k_srcMaskLenMask |
   k_dstMaskLenMask |
   k_tcpFlagsMask |
   k_engineTypeMask |
   k_engineIdMask;

# Not handling these currently!:
#   k_inputEncapMask |
#   k_outputEncapMask |
#   k_peerNextHopMask |

$Cflow::index = $Cflow::expected_entry_mask;

$Cflow::entry = '';
$Cflow::entry .= ' N'; # _index
$Cflow::entry .= ' N'; # _router
$Cflow::entry .= ' N'; # _srcIpAddr
$Cflow::entry .= ' N'; # _dstIpAddr
$Cflow::entry .= ' n'; # _inputIfIndex
$Cflow::entry .= ' n'; # _outputIfIndex
$Cflow::entry .= ' n'; # _srcPort
$Cflow::entry .= ' n'; # _dstPort
$Cflow::entry .= ' N'; # _pkts
$Cflow::entry .= ' N'; # _bytes
$Cflow::entry .= ' N'; # _ipNextHop
$Cflow::entry .= ' N'; # _startTime
$Cflow::entry .= ' N'; # _endTime
$Cflow::entry .= ' C'; # _protocol
$Cflow::entry .= ' C'; # _tos
$Cflow::entry .= ' n'; # _srcAs
$Cflow::entry .= ' n'; # _dstAs
$Cflow::entry .= ' C'; # _srcMaskLen
$Cflow::entry .= ' C'; # _dstMaskLen
$Cflow::entry .= ' C'; # _tcpFlags
if (k_inputEncapMask & $Cflow::index) {
   $Cflow::entry .= ' C'; # _inputEncap
}
if (k_outputEncapMask & $Cflow::index) {
   $Cflow::entry .= ' C'; # _outputEncap
}
if (k_peerNextHopMask & $Cflow::index) {
   $Cflow::entry .= ' N'; # _peerNextHop
}
$Cflow::entry .= ' C'; # _engineType
$Cflow::entry .= ' C'; # _engineId
$Cflow::entry_len = length(pack($Cflow::entry));

# As we go, we'll build an associative array of cached
# formats for the variable portion.  After a bit of time
# this should speed things up because we can do a direct
# lookup from the $Cflow::index to the format rather than
# having to construct the format from scratch each time.
%Cflow::cached_formats = ();

# tie these scalars so that we don't have to call some functions (such as
# strftime(3) and inet_ntoa(3)) unnecessarily.  (As tied scalars those
# functions will only be called if the values of these variables are actually
# fetched - i.e. if they're referred to by the callers "wanted" subroutine.
# This speeds things up (when possible) by saving a bit of processing time
# with each flow, which can add up to quite a bit for lots of flows.):
die unless tie($Cflow::localtime, 'Cflow::LocalTime', \$Cflow::endtime);
die unless tie($Cflow::exporterip, 'Cflow::InetNtoA', \$Cflow::exporter);
die unless tie($Cflow::srcip, 'Cflow::InetNtoA', \$Cflow::srcaddr);
die unless tie($Cflow::dstip, 'Cflow::InetNtoA', \$Cflow::dstaddr);
die unless tie($Cflow::nexthopip, 'Cflow::InetNtoA', \$Cflow::nexthop);
# Bytes per second, Packets per second:
die unless tie($Cflow::Bps, 'Cflow::PerSecond', \$Cflow::bytes);
die unless tie($Cflow::pps, 'Cflow::PerSecond', \$Cflow::pkts);
# TCPFlags:
die unless tie($Cflow::TCPFlags, 'Cflow::SymbolicTCPFlags', \$Cflow::tcp_flags);
# ICMPTypeCode:
die unless tie($Cflow::ICMPTypeCode, 'Cflow::SymbolicICMPTypeCode', \$Cflow::dstport);

sub find {
 my($ARGV);
 my($wanted, $perfile);
 my $total = 0;
 my $count = 0;
 if ('CODE' eq ref($_[0])) {
    $wanted = shift
 } else {
    $wanted = \&Cflow::wanted # a default "wanted" subroutine
 }
 if ('CODE' eq ref($_[0])) {
    $perfile = shift
 }

 FILE: foreach $ARGV (@_) {
  open(FLOWS, "<$ARGV") || die "open \"$ARGV\": $!\n";

  &$perfile($ARGV) if $perfile; # call the per-file "hook" if one was supplied.

  while (1) {
      last unless read(FLOWS, $Cflow::raw, $Cflow::entry_len);
      my($input_encap, $output_encap,
	 $peerNextHop,
	 $engine_type, $engine_id);
      ($Cflow::index, $Cflow::exporter,
       $Cflow::srcaddr, $Cflow::dstaddr,
       $Cflow::input_if, $Cflow::output_if,
       $Cflow::srcport, $Cflow::dstport,
       $Cflow::pkts, $Cflow::bytes,
       $Cflow::nexthop,
       $Cflow::startime, $Cflow::endtime,
       $Cflow::protocol, $Cflow::tos,
       $Cflow::src_as, $Cflow::dst_as,
       $Cflow::src_mask, $Cflow::dst_mask,
       $Cflow::tcp_flags,
       $engine_type, $engine_id) = unpack($Cflow::entry, $Cflow::raw);

      if ($Cflow::index != $Cflow::expected_entry_mask) {
         warn "$ARGV: Invalid index in flow data file: ", sprintf("%X", $Cflow::index), "!  Version 5 flow-export is required with *all* data being saved using the COLLECT field of the CISCOEXPORTER stanza(s)!\n" if $Cflow::verbose;
	 next FILE
      }
      $total++;

      $Cflow::unix_secs = $Cflow::endtime; # FIXME? Should this be $startime?

      if (1 == $Cflow::protocol) {
         $Cflow::ICMPType = $Cflow::dstport >> 8;
         $Cflow::ICMPCode = 0xff & $Cflow::dstport
      }

      # validate flow here...
      &$wanted && ++$count;
  }
 } continue {
  close(FLOWS);
 }
 return((0+$count) . '/' . (0+$total))
}

sub verbose {
   $Cflow::verbose = $_[0]
}

sub wanted {
   return if ($Cflow::srcport < 1024 || $Cflow::dstport < 1024);
   return unless ((($Cflow::srcport == 31337 || $Cflow::dstport == 31337) &&
		   17 == $Cflow::protocol) ||
		  (($Cflow::srcport == 12345 || $Cflow::srcport == 12346 ||
		    $Cflow::dstport == 12345 || $Cflow::dstport == 12346) &&
		   6 == $Cflow::protocol));

   my $when = POSIX::strftime("%Y/%m/%d %H:%M:%S", localtime($Cflow::unix_secs));

   printf "$when %15.15s.%-5hu %15.15s.%-5hu %2hu %10u %10u\n", $Cflow::srcip, $Cflow::srcport, $Cflow::dstip, $Cflow::dstport, $Cflow::protocol, $Cflow::pkts, $Cflow::bytes
}

1;