#  Cflow.pm - perl module for processing raw flow files
#  Copyright (C) 1998-2005  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.53 2005/09/28 15:58:01 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_URG
);

# 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 '' unless defined $typecode[$Cflow::ICMPType]->[0];
   return $typecode[$Cflow::ICMPType]->[0]
      unless defined $typecode[$Cflow::ICMPType]->[1]->[$Cflow::ICMPCode];
   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::duration_secs}.${Cflow::duration_msecs}";
   $$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::duration_secs}.${Cflow::duration_msecs}";
   $$this / (($seconds > 0.)? $seconds : 1)
}

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

package Cflow::ReRaw; ##########################################################
use Carp;

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

sub FETCH {
   my $this = shift;
   pack($Cflow::entry,
        $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,
        $Cflow::engine_type, $Cflow::engine_id);
}

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

package Cflow::KludgeSecs; #####################################################
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;
   ($$$this & 0xffff0000) >> 16
}

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

package Cflow::KludgeMsecs; ####################################################
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;
   $$$this & 0xffff
}

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.53 $' =~ m/(\d+)\.(\d+)/ && (( $Cflow::VERSION ) = sprintf("%d.%03d", $1, $2));

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

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

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));

bootstrap Cflow $Cflow::VERSION;

=head1 NAME

Cflow::find - find "interesting" flows in raw IP flow files

=head1 SYNOPSIS

   use Cflow;

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

   sub wanted { ... }

or:

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

   sub perfile { 
      my $fname = shift;
      ...
   }

=head1 BACKROUND

This module implements an API for processing IP flow accounting
information which as been collected from routers and written into flow
files by one of the various flow collectors listed below.

It was originally conceived and written for use by FlowScan:

   http://net.doit.wisc.edu/~plonka/FlowScan/

=head1 Flow File Sources

This package is of little use on its own.  It requires input in the
form of time-stamped raw flow files produced by other software
packages.  These "flow sources" either snoop a local ethernet (via
libpcap) or collect flow information from IP routers that are
configured to export said information.  The following flow sources are
supported:

=over 4

=item argus by Carter Bullard:

   http://www.qosient.com/argus/

=item flow-tools by Mark Fullmer (with NetFlow v1, v5, v6, or v7):

   http://www.splintered.net/sw/flow-tools/

=item CAIDA's cflowd (with NetFlow v5):

   http://www.caida.org/tools/measurement/cflowd/
   http://net.doit.wisc.edu/~plonka/cflowd/

=item lfapd by Steve Premeau (with LFAPv4):

   http://www.nmops.org/

=back

=head1 DESCRIPTION

Cflow::find() will iterate across all the flows in the specified
files.  It will call your wanted() function once per flow record.  If
the file name argument passed to find() is specified as "-", flows will
be read from standard input.

The wanted() function does whatever you write it to do.  For instance,
it could simply print interesting flows or it might maintain byte,
packet, and flow counters which could be written to a database after
the find subroutine completes.

Within your wanted() function, tests on the "current" flow can be
performed using the following variables:

=over 4

=item $Cflow::unix_secs

secs since epoch (deprecated)

=item $Cflow::exporter

Exporter IP Address as a host-ordered "long"

=item $Cflow::exporterip

Exporter IP Address as dotted-decimal string

=item $Cflow::localtime

$Cflow::unix_secs interpreted as localtime with this strftime(3) format:

   %Y/%m/%d %H:%M:%S

=item $Cflow::srcaddr

Source IP Address as a host-ordered "long"

=item $Cflow::srcip

Source IP Address as a dotted-decimal string

=item $Cflow::dstaddr

Destination IP Address as a host-ordered "long"

=item $Cflow::dstip

Destination IP Address as a dotted-decimal string

=item $Cflow::input_if

Input interface index

=item $Cflow::output_if

Output interface index

=item $Cflow::srcport

TCP/UDP src port number or equivalent

=item $Cflow::dstport

TCP/UDP dst port number or equivalent

=item $Cflow::ICMPType

high byte of $Cflow::dstport

Undefined if the current flow is not an ICMP flow.

=item $Cflow::ICMPCode

low byte of $Cflow::dstport

Undefined if the current flow is not an ICMP flow.

=item $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
   HOST_UNREACH

Undefined if the current flow is not an ICMP flow.

=item $Cflow::pkts

Packets sent in Duration

=item $Cflow::bytes

Octets sent in Duration

=item $Cflow::nexthop

Next hop router's IP Address as a host-ordered "long"

=item $Cflow::nexthopip

Next hop router's IP Address as a dotted-decimal string

=item $Cflow::startime

secs since epoch at start of flow

=item $Cflow::start_msecs

fractional portion of startime (in milliseconds)

This will be zero unless the source is flow-tools or argus.

=item $Cflow::endtime

secs since epoch at last packet of flow

=item $Cflow::end_msecs

fractional portion of endtime (in milliseconds)

This will be zero unless the source is flow-tools or argus.

=item $Cflow::protocol

IP protocol number (as is specified in F</etc/protocols>, i.e.
1=ICMP, 6=TCP, 17=UDP, etc.)

=item $Cflow::tos

IP Type-of-Service

=item $Cflow::tcp_flags

bitwise OR of all TCP flags that were set within packets in the flow;
0x10 for non-TCP flows

=item $Cflow::TCPFlags

symbolic representation of $Cflow::tcp_flags
The value will be a bitwise-or expression.
E.g.

   PUSH|SYN|FIN|ACK

Undefined if the current flow is not a TCP flow.

=item $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. E.g.:

   syswrite(FILEHANDLE, $Cflow::raw, length $Cflow::raw)

Note that if you're using a modern version of perl that supports PerlIO
Layers, be sure that FILEHANDLE is using something appropriate like the
":bytes" layer.  This can be activated on open, or with:

   binmode(FILEHANDLE, ":bytes");

This will prevent the external LANG setting from causing perl to do
such things as interpreting your raw flow records as UTF-8 characters
and corrupting the record.

=item $Cflow::reraw

the entire "re-packed" flow record formatted like $Cflow::raw.

This is useful when the "wanted" subroutine wants to write a modified flow
to another FILEHANDLE. E.g.:

   $srcaddr = my_encode($srcaddr);
   $dstaddr = my_encode($dstaddr);
   syswrite(FILEHANDLE, $Cflow::reraw, length $Cflow::raw)

These flow variables are packed into $Cflow::reraw:

   $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,
   $Cflow::engine_type, $Cflow::engine_id

Note that if you're using a modern version of perl that supports PerlIO
Layers, be sure that FILEHANDLE is using something appropriate like the
":bytes" layer.  This can be activated on open, or with:

   binmode(FILEHANDLE, ":bytes");

This will prevent the external LANG setting from causing perl to do
such things as interpreting your raw flow records as UTF-8 characters
and corrupting the record.

=item $Cflow::Bps

the minimum bytes per second for the current flow

=item $Cflow::pps

the minimum packets per second for the current flow

=back

The following variables are undefined if using NetFlow v1 (which does
not contain the requisite information):

=over 4

=item $Cflow::src_as

originating or peer AS of source address

=item $Cflow::dst_as

originating or peer AS of destination address

=back

The following variables are undefined if using NetFlow v1 or LFAPv4
(which do not contain the requisite information):

=over 4

=item $Cflow::src_mask

source address prefix mask bits

=item $Cflow::dst_mask

destination address prefix mask bits

=item $Cflow::engine_type

type of flow switching engine

=item $Cflow::engine_id

ID of the flow switching engine

=back

Optionally, a reference to a perfile() function can be passed to
Cflow::find 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 the flow
file which is about to be processed.  The purpose of the perfile()
function is to allow you to periodically report the progress of
Cflow::find() and to provide an opportunity to periodically reclaim
storage used by data objects that may have been allocated or maintained
by the wanted() function.  For instance, when counting the number of
active hosts IP addresses in each time-stamped flow file, perfile() can
reset the counter to zero and clear the search tree or hash used to
remember those IP addresses.

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.031);

Cflow::find() returns a "hit-ratio".  This hit-ratio is a string
formatted similarly to that of the value of a perl hash when taken in a
scalar context.  This hit-ratio indicates ((# of "wanted" flows) / (#
of scanned flows)).  A flow is considered to have been "wanted" if your
wanted() function returns non-zero.

Cflow::verbose() takes a single scalar boolean argument which indicates
whether or not you wish warning messages to be generated to STDERR when
"problems" occur.  Verbose mode is set by default.

=head1 EXAMPLES

Here's a complete example with a sample wanted function.  It will print
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);

   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)
   }

Here's an example which demonstrates a technique which can be used to
pass arbitrary arguments to your wanted function by passing a reference
to an anonymous subroutine as the wanted() function argument to
Cflow::find():

   sub wanted {
      my @params = @_;
      # ...
   }

   Cflow::find(sub { wanted(@params) }, @files);

=head1 ARGUS NOTES

Argus uses a bidirectional flow model.  This means that some argus
flows represent packets not only in the forward direction (from
"source" to "destination"), but also in the reverse direction (from the
so-called "destination" to the "source").  However, this module uses a
unidirection flow model, and therfore splits some argus flows into two
unidirectional flows for the purpose of reporting.

Currently, using this module's API there is no way to determine if two
subsequently reported unidirectional flows were really a single argus
flow.  This may be addressed in a future release of this package.

Furthermore, for argus flows which represent bidirectional ICMP
traffic, this module presumes that all the reverse packets were
ECHOREPLYs (sic).  This is sometimes incorrect as described here:

   http://www.theorygroup.com/Archive/Argus/2002/msg00016.html

and will be fixed in a future release of this package.

Timestamps ($startime and $endtime) are sometimes reported incorrectly
for bidirectional argus flows that represent only one packet in each
direction.  This will be fixed in a future release.

Argus flows sometimes contain information which does not map directly
to the flow variables presented by this module.  For the time being,
this information is simply not accessible through this module's API.
This may be addressed in a future release.

Lastly, argus flows produced from observed traffic on a local ethernet
do not contain enough information to meaningfully set the values of all
this module's flow variables.  For instance, the next-hop and
input/output ifIndex numbers are missing.  For the time being, all
argus flows accessed throught this module's API will have both the
$input_if and $output_if as 42.  Althought 42 is the answer to life,
the universe, and everthing, in this context, it is just an arbitrary
number.  It is important that $output_if is non-zero, however, since
existing FlowScan reports interpret an $output_if value of zero to mean
that the traffic represented by that flow was not forwarded (i.e.
dropped).  For similar reasons, the $nexthopip for all argus flows is
reported as "127.0.0.1".

=head1 BUGS

Currently, only NetFlow version 5 is supported when reading
cflowd-format raw flow files.

When built with support for flow-tools and attempting to read
a cflowd format raw flow file from standard input, you'll get the
error:

   open "-": No such file or directory

For the time being, the workaround is to write the content to a file
and read it from directly from there rather than from standard input.
(This happens because we can't close and re-open file descriptor zero
after determining that the content was not in flow-tools format.)

When built with support for flow-tools and using verbose mode,
Cflow::find will generate warnings if you process a cflowd format raw
flow file.  This happens because it will first attempt to open the file
as a flow-tools format raw flow file (which will produce a warning
message), and then revert to handling it as cflowd format raw flow
file.

Likewise, when built with support for argus and attempting to read a
cflowd format raw flow file from standard input, you'll get this
warning message:

   not Argus-2.0 data stream.

This is because argus (as of argus-2.0.4) doesn't seem to have a mode
in which such warning messages are supressed.

The $Cflow::raw flow variable contains the flow record in cflowd
format, even if it was read from a raw flow file produced by flow-tools
or argus.  Because cflowd discards the fractional portion of the flow
start and end time, only the whole seconds portion of these times will
be retained.  (That is, the raw record in $Cflow::raw does not contain
the $start_msecs and $end_msecs, so using $Cflow::raw to convert to
cflowd format is a lossy operation.)

When used with cflowd, Cflow::find() will generate warnings if the flow
data file is "invalid" as far as its concerned.  To avoid this, you
must be using Cisco version 5 flow-export and configure cflowd so that
it saves all flow-export data.  This is the default behavior when
cflowd produces time-stamped raw flow files after being patched as
described here:

   http://net.doit.wisc.edu/~plonka/cflowd/

=head1 NOTES

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

=head1 AUTHOR

Dave Plonka <plonka@doit.wisc.edu>

Copyright (C) 1998-2005  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.53 $>)
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.

=head1 SEE ALSO

perl(1), Socket, Net::Netmask, Net::Patricia.

=cut

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

$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
$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);

die unless tie($Cflow::reraw, 'Cflow::ReRaw');

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;
__END__