File Coverage

blib/lib/Net/Flow.pm

Criterion	Covered	Total	%
statement	162	449	36.0
branch	33	190	17.3
condition	3	44	6.8
subroutine	24	30	80.0
pod	2	11	18.1
total	224	724	30.9

#!/usr/bin/perl
#
#
# Atsushi Kobayashi
#
# Acknowledgments
# This module was supported by the Ministry of Internal Affairs and
# Communications of Japan.
#
# Flow.pm - 2008/12/04
#
# Copyright (c) 2007-2008 NTT Information Sharing Platform Laboratories
#
# This package is free software and is provided "as is" without express
# or implied warranty.  It may be used, redistributed and/or modified
# under the terms of the Perl Artistic License (see
# http://www.perl.com/perl/misc/Artistic.html)
#

package Net::Flow;

use 5.008008;
use strict;
use warnings;

use Time::HiRes qw(tv_interval gettimeofday);
use Digest::MD5 qw(md5_hex);
use Data::Dumper;

use Exporter;

our @EXPORT_OK = qw(decode encode);
our $VERSION   = '1.003';

use constant NetFlowv5 => 5;
use constant NetFlowv8 => 8;
use constant NetFlowv9 => 9;
use constant IPFIX     => 10;

use constant NFWV9_DataTemplateSetId   => 0;
use constant NFWV9_OptionTemplateSetId => 1;

use constant IPFIX_DataTemplateSetId   => 2;
use constant IPFIX_OptionTemplateSetId => 3;

use constant MinDataSetId        => 256;
use constant VariableLength      => 65535;
use constant ShortVariableLength => 255;

my %TemplateForNetFlowv5 = (
	'SetId'      => 0,
	'TemplateId' => 0,
	'FieldCount' => 20,
	'Template'   => [
		{ 'Length' => 4,	'Id' => 8  }, # sourceIPv4Address
		{ 'Length' => 4,	'Id' => 12 }, # destinationIPv4Address
		{ 'Length' => 4,	'Id' => 15 }, # ipNextHopIPv4Address
		{ 'Length' => 2,	'Id' => 10 }, # ingressInterface
		{ 'Length' => 2,	'Id' => 14 }, # egressInterface
		{ 'Length' => 4,	'Id' => 2  }, # packetDeltaCount
		{ 'Length' => 4,	'Id' => 1  }, # octetDeltaCount
		{ 'Length' => 4,	'Id' => 22 }, # flowStartSysUpTime
		{ 'Length' => 4,	'Id' => 21 }, # flowEndSysUpTime
		{ 'Length' => 2,	'Id' => 7  }, # sourceTransportPort
		{ 'Length' => 2,	'Id' => 11 }, # destinationTransportPort
		{ 'Length' => 1,	'Id' => 210}, # paddingOctets
		{ 'Length' => 1,	'Id' => 6  }, # tcpControlBits
		{ 'Length' => 1,	'Id' => 4  }, # protocolIdentifier
		{ 'Length' => 1,	'Id' => 5  }, # ipClassOfService
		{ 'Length' => 2,	'Id' => 16 }, # bgpSourceAsNumber
		{ 'Length' => 2,	'Id' => 17 }, # bgpDestinationAsNumber
		{ 'Length' => 1,	'Id' => 9  }, # sourceIPv4PrefixLength
		{ 'Length' => 1,	'Id' => 13 }, # destinationIPv4PrefixLength
		{ 'Length' => 2,	'Id' => 210}, # paddingOctets
	],
);

#################### START sub encode() ####################
sub encode {

	my ( $InputHeaderRef, $InputTemplateRef, $InputFlowRef, $MaxDatagram ) = @_;
	my @Payloads            = ();
	my @FlowPacks           = ();
	my %FlowSetPayloads     = ();
	my $FlowSetHeaderLength = 4;
	my @Errors              = ();
	my $Error               = undef;

	# This is a terrible default, but it was the original behavior.
	$InputHeaderRef->{TemplateResendSecs} = 0
		unless defined $InputHeaderRef->{TemplateResendSecs};

	$InputHeaderRef->{_sysstarttime} ||= [gettimeofday];
  check_header($InputHeaderRef) unless defined $InputHeaderRef->{_header_len};

	my $sendTemplates = 1;    # Always is the default

	my $template_info;

	# if TemplateResendSecs is true someone has bothered to define it so
	# we can do some extra work to see if we really need to send
	# template info.
	if ( $InputHeaderRef->{TemplateResendSecs} ) {
		my $templates_id = scalar $InputTemplateRef;
		##warn "templates_id: $templates_id\n";
		$InputHeaderRef->{_template_info}->{$templates_id} ||= {};
		$template_info = $InputHeaderRef->{_template_info}->{$templates_id};
		my $hash = md5_hex( Dumper($InputTemplateRef) );
		##warn Dumper($InputTemplateRef) unless defined $template_info->{hash};
		$template_info->{hash} ||= $hash;
		if ( $template_info->{hash} ne $hash ) {
			##warn "$template_info->{hash} ne $hash", "\n";
			##warn Dumper($InputTemplateRef);
			$template_info->{hash} = $hash;
			delete $template_info->{_template_sent};
		}

		# This is a kludge until I come up with something better.  Using
		# the stringified $InputTemplateRef as an ID works, but if someone
		# is passing in a new ref everytime this will slowly leak memory.
		#
		# I have arbitrarily chosen 50 as too many sets of template
		# information to have.  Hopefully this will keep the amount of
		# looping through the _template_info hash to a minimum and also
		# prevent memory from leaking endlessly.
		if ( keys %{ $InputHeaderRef->{_template_info} } > 50 ) {
			for my $key ( keys %{ $InputHeaderRef->{_template_info} } ) {
				my $sent = $InputHeaderRef->{_template_info}->{$key}->{_template_sent};
				if ( time - $sent > $InputHeaderRef->{TemplateResendSecs} ) {
					delete $InputHeaderRef->{_template_info}->{$key};
				}
			}
		}

		$sendTemplates = (
			( !defined $template_info->{_template_sent} )
			? 1
			: ( ( time - $template_info->{_template_sent} ) >= $InputHeaderRef->{TemplateResendSecs} )
		);
	}

	#
	# check header reference
	#

	my ($ErrorRef) = &check_header($InputHeaderRef);

	push( @Errors, @{$ErrorRef} ) if ( defined $ErrorRef );

	my @flowRef;
	if ($sendTemplates) {
		push @flowRef, @{$InputTemplateRef};
		$template_info->{_template_sent} = time if ref $template_info eq 'HASH';
	}
	push @flowRef, @{$InputFlowRef};

	##warn scalar localtime ($template_info->{_template_sent}), "\n";

	foreach my $FlowRef (@flowRef) {
		my $PackRef           = undef;
		my $ErrorRef          = undef;
		my $DecodeTemplateRef = undef;

		unless ( defined $FlowRef->{SetId} ) {
			$Error = 'ERROR : NOTHING SETID VALUE';
			push( @Errors, $Error );
			next;
		}

		#
		# encode flow data
		#

		if ( $FlowRef->{SetId} >= MinDataSetId ) {

			#
			# searching for particular template
			#

			( $DecodeTemplateRef, $Error ) = &search_template( $FlowRef->{SetId}, $InputTemplateRef );

			if ( defined $DecodeTemplateRef ) {

				( $PackRef, $ErrorRef ) = &flow_encode( $FlowRef, $DecodeTemplateRef );

			} else {

				$Error = "ERROR : NO TEMPLATE TEMPLATE ID=$FlowRef->{SetId}";
				push( @Errors, $Error );

			}

			#
			# encode template data
			#

		} else {

			( $PackRef, $ErrorRef ) = &template_encode( $FlowRef, $InputHeaderRef );

		}

		push( @FlowPacks, $PackRef )
			if defined $PackRef;

		push( @Errors, @{$ErrorRef} ) if defined $ErrorRef;

	}

	unless (@FlowPacks) {

		$Error = 'ERROR : NO FLOW DATA';
		push( @Errors, $Error );
		return ( $InputHeaderRef, \@Payloads, \@Errors );

	}

	#
	# encode NetFlowv9/IPFIX datagram
	#

	my $FlowCount = 0;
	my $DataCount = 0;
	foreach my $FlowPackRef (@FlowPacks) {

		unless ( defined $FlowPackRef->{Pack} ) {
			warn 'undefined $FlowPackRef->{Pack}', "\n";
			next;
		}

		#
		# check datagram size
		#

		my $TotalLength = $InputHeaderRef->{_header_len};

		foreach my $SetId ( keys %FlowSetPayloads ) {

			$TotalLength += length( $FlowSetPayloads{$SetId} ) + $FlowSetHeaderLength;

		}

		if ( ( length( $FlowPackRef->{Pack} ) + $TotalLength ) > $MaxDatagram ) {

			#
			# make NetFlow/IPFIX datagram
			#

			push( @Payloads, &datagram_encode( $InputHeaderRef, \%FlowSetPayloads, \$FlowCount, \$DataCount ) );

			%FlowSetPayloads = ();
			$FlowCount       = 0;
			$DataCount       = 0;
		}

		$FlowSetPayloads{ $FlowPackRef->{SetId} } .= $FlowPackRef->{Pack};

		$DataCount++ if $FlowPackRef->{SetId} >= MinDataSetId;
		$FlowCount++;

	}

	# Push a final flow if any.
	if ( $FlowCount > 0 ) {
		push( @Payloads, &datagram_encode( $InputHeaderRef, \%FlowSetPayloads, \$FlowCount, \$DataCount ) );

	}

	return ( $InputHeaderRef, \@Payloads, \@Errors );

}
#################### END sub encode() ######################

#################### START sub check_header() ##############

sub check_header {
	my ($InputHeaderRef) = @_;

	my @Errors;
	my $Error;

	if ( $InputHeaderRef->{VersionNum} == IPFIX ) {
		$InputHeaderRef->{_header_len} = 16;
		$InputHeaderRef->{ObservationDomainId} ||= 0;
		$InputHeaderRef->{SequenceNum}         ||= 0;
		$InputHeaderRef->{_export_time} = $InputHeaderRef->{UnixSecs} || time;
	} else {
		if ( $InputHeaderRef->{VersionNum} != NetFlowv9 ) {
			if ( !defined $InputHeaderRef->{VersionNum} ) {
				$Error = 'WARNING : NO HEADER VERSION NUMBER';
			} else {
				$Error = "WARNING : NO SUPPORT HEADER VERSION NUMBER $InputHeaderRef->{VersionNum}";
			}
      push( @Errors, $Error );
		}
		$InputHeaderRef->{VersionNum} = NetFlowv9;

		$InputHeaderRef->{_header_len} = 20;
		$InputHeaderRef->{SourceId}    ||= 0;
		$InputHeaderRef->{SequenceNum} ||= 0;
		$InputHeaderRef->{_export_time} = $InputHeaderRef->{UnixSecs} || time;
		$InputHeaderRef->{SysUpTime} = int( tv_interval( $InputHeaderRef->{_sysstarttime} ) * 1000 );
	}

	return ( \@Errors );
}
#################### END sub check_header() ################

#################### START sub datagram_encode() ###########
sub datagram_encode {
	my ( $HeaderRef, $FlowSetPayloadRef, $FlowCountRef, $DataCountRef ) = @_;
	my $Payload = '';

	#
	# encode flow set data
	#

	foreach my $SetId ( sort { $a <=> $b } ( keys %{$FlowSetPayloadRef} ) ) {

		#
		# make padding part
		#

		my $padding = '';

		while ( ( length( $FlowSetPayloadRef->{$SetId} ) + length($padding) ) % 4 != 0 ) {
			$padding .= "\0";
		}

		my $set_len = ( length( $FlowSetPayloadRef->{$SetId} ) + length($padding) + 4 );

		# Pack set header
		$Payload .= pack( 'n2', $SetId, $set_len );

		# Pack set data
		$Payload .= $FlowSetPayloadRef->{$SetId};

		# Pack padding
		$Payload .= $padding;
	}

	if ( $HeaderRef->{VersionNum} == NetFlowv9 ) {

    $HeaderRef->{SysUpTime} ||= 0;
    $HeaderRef->{_export_time} ||= 0;
    $HeaderRef->{SequenceNum} ||= 0;
		$HeaderRef->{SequenceNum} = ( $HeaderRef->{SequenceNum} + 1 ) % 0xFFFFFFFF;
		$HeaderRef->{Count}       = $$FlowCountRef;

		$Payload = pack( 'n2N4', @{$HeaderRef}{qw{VersionNum Count SysUpTime _export_time SequenceNum SourceId}} ) . $Payload;

	} elsif ( $HeaderRef->{VersionNum} == IPFIX ) {

		$Payload = pack( 'n2N3', $HeaderRef->{VersionNum}, ( length($Payload) + $HeaderRef->{_header_len} ), @{$HeaderRef}{qw{_export_time SequenceNum ObservationDomainId}} ) . $Payload;

		$HeaderRef->{SequenceNum} = ( $HeaderRef->{SequenceNum} + $$DataCountRef ) % 0xFFFFFFFF;

	}


	return ( \$Payload );

}
#################### END sub datagram_encode() #############

#################### START sub flow_encode() ###############
sub flow_encode {
	my ( $FlowRef, $DecodeTemplateRef ) = @_;
	my %FlowData = ();
	my @Errors   = ();
	my $Error    = undef;
	my %Count    = ();

	$FlowData{SetId} = $DecodeTemplateRef->{TemplateId};

	foreach my $TemplateArrayRef ( @{ $DecodeTemplateRef->{Template} } ) {

		my $FlowValue = undef;


		$Count{ $TemplateArrayRef->{Id} } ||= 0;

		if ( defined $FlowRef->{ $TemplateArrayRef->{Id} } ) {

			#
			# One Template has multiple same Ids.
			#

			if ( ref $FlowRef->{ $TemplateArrayRef->{Id} } ) {

				$FlowValue = @{ $FlowRef->{ $TemplateArrayRef->{Id} } }[ $Count{ $TemplateArrayRef->{Id} } ];

				#
				# Each Id is different than others.
				#

			} else {

				$FlowValue = $FlowRef->{ $TemplateArrayRef->{Id} };

			}

			#
			# Variable Length Type
			#

			if ( $TemplateArrayRef->{Length} == VariableLength ) {

				my $Length = length($FlowValue);

				#
				# Value Length  < 255
				#

				if ( $Length < ShortVariableLength ) {

					$FlowData{Pack} .= pack( "C A$Length", $Length, $FlowValue );

					#
					# Value Length > 255
					#

				} else {

					$FlowData{Pack} .= pack( "C n A$Length", ShortVariableLength, $Length, $FlowValue );

				}

				#
				# Fixed Length Type
				#

			} else {

				$FlowData{Pack} .= pack( "A$TemplateArrayRef->{Length}", $FlowValue );

			}


		} else {
			$Data::Dumper::Sortkeys = sub {
				my $h = shift;
				return [
					sort {
						if ( $a =~ /^\d+$/ && $b =~ /^\d+$/ ) {
							$a <=> $b;
						} else {
							lc($a) cmp lc($b);
						}
					} ( keys %$h )
				];
			};

			$Error = "WARNING : NOT FIELD DATA INFORMATION ELEMENT ID=$TemplateArrayRef->{Id}";
			push( @Errors, $Error );

			if ( $TemplateArrayRef->{Length} == VariableLength ) {

				$FlowData{Pack} .= pack( 'C', 0 );

			} else {

				$FlowData{Pack} .= pack("a$TemplateArrayRef->{Length}");

			}

		}

		$Count{ $TemplateArrayRef->{Id} }++;

	}

	return ( \%FlowData, \@Errors );

}
#################### END sub flow_encode() #################

#################### START sub template_encode() ###########
sub template_encode {
	my ( $TemplateRef, $HeaderRef ) = @_;
	my %TemplateData = ();
	my $ScopeCount   = 0;
	my @Errors       = ();
	my $Error        = undef;

	#
	# check template hash reference
	#

	unless ( defined $TemplateRef->{TemplateId} ) {
		$Error = 'ERROR : NO TEMPLATE ID';
		push( @Errors, $Error );
	}

	unless ( defined $TemplateRef->{SetId} ) {
		$Error = 'ERROR : NO SET ID';
		push( @Errors, $Error );
	}

	if ( $HeaderRef->{VersionNum} == NetFlowv9 ) {

		if (   $TemplateRef->{SetId} != NFWV9_DataTemplateSetId
			&& $TemplateRef->{SetId} != NFWV9_OptionTemplateSetId ) {

			$Error = "ERROR : UNMATCH SET ID FOR NETFLOWV9 TEMPLATE=$TemplateRef->{TemplateId}";
			push( @Errors, $Error );

		}

	} elsif ( $HeaderRef->{VersionNum} == IPFIX ) {

		if (   $TemplateRef->{SetId} != IPFIX_DataTemplateSetId
			&& $TemplateRef->{SetId} != IPFIX_OptionTemplateSetId ) {

			$Error = "ERROR : UNMATCH SET ID FOR IPFIX TEMPLATE=$TemplateRef->{TemplateId}";
			push( @Errors, $Error );

		}

	}

	return ( \%TemplateData, \@Errors ) if $#Errors >= 0;

	$TemplateData{SetId} = $TemplateRef->{SetId};

	$ScopeCount = $TemplateRef->{ScopeCount}
		if defined $TemplateRef->{ScopeCount};

	$TemplateRef->{FieldCount} = $#{ $TemplateRef->{Template} } + 1
		unless defined $TemplateRef->{FieldCount};

	#
	# NetFlow v9 pack data template header
	#

	if ( $TemplateRef->{SetId} == NFWV9_DataTemplateSetId ) {

		$TemplateData{Pack} = pack( 'n2', @{$TemplateRef}{qw{TemplateId FieldCount}} );

		#
		# NetFlow v9 pack option template header
		#

	} elsif ( $TemplateRef->{SetId} == NFWV9_OptionTemplateSetId ) {

		$TemplateData{Pack} = pack( 'n3', $TemplateRef->{TemplateId}, $ScopeCount * 4, ( $#{ $TemplateRef->{Template} } + 1 - $ScopeCount ) * 4, );

		#
		# IPFIX pack data template header
		#

	} elsif ( $TemplateRef->{SetId} == IPFIX_DataTemplateSetId ) {

		#
		# Template Withdraw
		#

		if ( $TemplateRef->{FieldCount} == 0 ) {

			$TemplateData{Pack} = pack( 'n2', $TemplateRef->{TemplateId}, 0 );

		} else {

			$TemplateData{Pack} = pack( 'n2', @{$TemplateRef}{qw{TemplateId FieldCount}} );

		}

		#
		# IPFIX pack option template header
		#

	} elsif ( $TemplateRef->{SetId} == IPFIX_OptionTemplateSetId ) {

		#
		# Template Withdraw
		#

		if ( $TemplateRef->{FieldCount} == 0 ) {

			$TemplateData{Pack} = pack( 'n2', $TemplateRef->{TemplateId}, 0 );

		} else {

			$TemplateData{Pack} = pack(
				'n3',
				$TemplateRef->{TemplateId},
				( $#{ $TemplateRef->{Template} } + 1 ),    # -$ScopeCount
				$ScopeCount,
			);
		}

	}

	#
	# pack template
	#

	if ( $TemplateRef->{FieldCount} > 0 ) {

		foreach my $Ref ( @{ $TemplateRef->{Template} } ) {

			#
			# Enterprise Num
			#

			if ( $Ref->{Id} =~ /([\d]+)\.([\d]+)/ ) {

				$TemplateData{Pack} .= pack( 'n2N', $2 + 0x8000, $Ref->{Length}, $1, );

			} else {

				$TemplateData{Pack} .= pack( 'n2', $Ref->{Id}, $Ref->{Length} );

			}

		}

	}

	return ( \%TemplateData, \@Errors );

}
#################### END sub template_encode() #############


#################### START sub decode() ####################
sub decode {
	my ( $NetFlowPktRef, $InputTemplateRef ) = @_;
	my $NetFlowHeaderRef = undef;
	my $FlowSetHeaderRef = undef;
	my $TemplateRef      = undef;
	my @Template         = ();
	my @Flows            = ();
	my @Errors           = ();
	my $Error            = undef;

	my $OffSet        = 0;
	my $FlowSetOffSet = 0;
	my $FlowCount     = 0;

	#
	# check packet data
	#

	if ( ref($NetFlowPktRef) ne 'SCALAR' ) {

		$Error = 'ERROR : NO PACKET DATA';
		push( @Errors, $Error );

		return ( $NetFlowHeaderRef, \@Template, \@Flows, \@Errors );

	}

	#
	# insert template data
	#

	if ( defined($InputTemplateRef) || ref($InputTemplateRef) eq 'ARRAY' ) {

		push( @Template, @{$InputTemplateRef} );

	} elsif ( defined($InputTemplateRef) ) {

		$Error = 'WARNING : NOT REF TEMPLATE DATA';
		push( @Errors, $Error );

	}

	#
	# header decode
	#

	( $NetFlowHeaderRef, $Error ) = &header_decode( $NetFlowPktRef, \$OffSet );

	#
	# IPFIX decode
	#

	if ( $NetFlowHeaderRef->{VersionNum} == IPFIX ) {

		while ( $OffSet < $NetFlowHeaderRef->{Length} ) {

			my $DecodeTemplateRef = undef;
			my $FlowRef           = undef;
			my $TemplateRef       = undef;

			if ( ( length($$NetFlowPktRef) - $OffSet ) < 4 ) {

				if ( $FlowCount ne $NetFlowHeaderRef->{Count} ) {
					$Error = 'WARNING : UNMATCH FLOW COUNT';
					push( @Errors, $Error );
				}

				last;
			}

			$FlowSetOffSet = $OffSet;

			#
			# decode flowset
			#

			$FlowSetHeaderRef = &flowset_decode( $NetFlowPktRef, \$OffSet );

			#
			# search for template
			#

			if ( $FlowSetHeaderRef->{SetId} >= MinDataSetId ) {

				( $DecodeTemplateRef, $Error ) = &search_template( $FlowSetHeaderRef->{SetId}, \@Template );

				unless ( defined $DecodeTemplateRef ) {

					push( @Errors, $Error );
					$OffSet = $FlowSetHeaderRef->{Length} + $FlowSetOffSet
						if defined $FlowSetHeaderRef->{Length};

					next;

				}

			}

			while ( $FlowSetHeaderRef->{Length} > ( $OffSet - $FlowSetOffSet ) ) {

				#
				# check word alignment
				#

				if ( ( $FlowSetHeaderRef->{Length} - ( $OffSet - $FlowSetOffSet ) ) < 4 ) {

					$OffSet = $FlowSetHeaderRef->{Length} + $FlowSetOffSet;
					last;

				}

				#
				# decode data template or option Template
				#

				if ( $FlowSetHeaderRef->{SetId} < MinDataSetId ) {

					( $TemplateRef, $Error ) = &template_decode( $NetFlowPktRef, \$OffSet, $FlowSetHeaderRef, \$NetFlowHeaderRef->{VersionNum} );

					if ( defined $Error ) {

						push( @Errors, $Error );
						last;

					}

					$FlowCount++;

					@Template = grep { $_ if ( $_->{TemplateId} ne $TemplateRef->{TemplateId} ); } @Template;

					push( @Template, $TemplateRef );

					#
					# decode flow records
					#

				} else {

					( $FlowRef, $Error ) = &flow_decode( $NetFlowPktRef, \$OffSet, $DecodeTemplateRef );

					if ( defined $Error ) {
						push( @Errors, $Error );
						last;
					}

					$FlowCount++;
					push( @Flows, $FlowRef );

				}

			}

		}

		#
		# NetFlow version 9 decode
		#

	} elsif ( $NetFlowHeaderRef->{VersionNum} == NetFlowv9 ) {

		while ( $FlowCount < $NetFlowHeaderRef->{Count} ) {
			my $DecodeTemplateRef = undef;
			my $FlowRef           = undef;
			my $TemplateRef       = undef;

			if ( ( length($$NetFlowPktRef) - $OffSet ) < 4 ) {

				if ( $FlowCount ne $NetFlowHeaderRef->{Count} ) {
					$Error = 'WARNING : UNMATCH FLOW COUNT';
					push( @Errors, $Error );
				}

				last;
			}

			$FlowSetOffSet = $OffSet;

			#
			# decode flowset
			#

			$FlowSetHeaderRef = &flowset_decode( $NetFlowPktRef, \$OffSet );

			#
			# search for template
			#

			if ( $FlowSetHeaderRef->{SetId} >= MinDataSetId ) {

				( $DecodeTemplateRef, $Error ) = &search_template( $FlowSetHeaderRef->{SetId}, \@Template );

				unless ( defined $DecodeTemplateRef ) {

					push( @Errors, $Error );
					$OffSet = $FlowSetHeaderRef->{Length} + $FlowSetOffSet
						if defined $FlowSetHeaderRef->{Length};

					next;

				}

			}

			while ( $FlowSetHeaderRef->{Length} > ( $OffSet - $FlowSetOffSet ) ) {

				#
				# check word alignment
				#

				if ( ( $FlowSetHeaderRef->{Length} - ( $OffSet - $FlowSetOffSet ) ) < 4 ) {

					$OffSet = $FlowSetHeaderRef->{Length} + $FlowSetOffSet;
					last;

				}

				#
				# decode data template or option Template
				#

				if ( $FlowSetHeaderRef->{SetId} < MinDataSetId ) {

					( $TemplateRef, $Error ) = &template_decode( $NetFlowPktRef, \$OffSet, $FlowSetHeaderRef, \$NetFlowHeaderRef->{VersionNum} );

					if ( defined $Error ) {

						push( @Errors, $Error );
						last;

					}

					$FlowCount++;

					@Template = grep { $_ if ( $_->{TemplateId} ne $TemplateRef->{TemplateId} ); } @Template;

					push( @Template, $TemplateRef );

					#
					# decode flow records
					#

				} else {

					( $FlowRef, $Error ) = &flow_decode( $NetFlowPktRef, \$OffSet, $DecodeTemplateRef );

					if ( defined $Error ) {
						push( @Errors, $Error );
						last;
					}

					$FlowCount++;
					push( @Flows, $FlowRef );

				}

			}

		}

		#
		# NetFlow version 5 Decode
		#

	} elsif ( $NetFlowHeaderRef->{VersionNum} == NetFlowv5 ) {

		while ( $FlowCount < $NetFlowHeaderRef->{Count} ) {

			my $FlowRef = undef;

			( $FlowRef, $Error ) = &flow_decode( $NetFlowPktRef, \$OffSet, \%TemplateForNetFlowv5 );

			$FlowRef->{SetId} = undef;

			if ( defined $Error ) {

				push( @Errors, $Error );
				last;

			}

			$FlowCount++;
			push( @Flows, $FlowRef );

		}

		#
		# NetFlow version 8 Decode
		#

	} elsif ( $NetFlowHeaderRef->{VersionNum} == NetFlowv8 ) {

		$Error = 'ERROR : NOT SUPPORT NETFLOW VER.8';
		push( @Errors, $Error );

	} else {

		$Error = 'ERROR : NOT NETFLOW DATA';
		push( @Errors, $Error );

	}

	return ( $NetFlowHeaderRef, \@Template, \@Flows, \@Errors );

}
#################### END sub decode() ######################

#################### START sub search_template() ###########
sub search_template {
	my ( $TemplateId, $TemplatesArrayRef ) = @_;
	my $DecodeTemplateRef = undef;
	my $Error             = undef;

	( $DecodeTemplateRef, undef ) = grep { $_ if $_->{TemplateId} eq $TemplateId; } @{$TemplatesArrayRef};

	#
	# nothing template for flow data
	#

	unless ( defined $DecodeTemplateRef ) {
		$Error = "WARNING : NOT FOUND TEMPLATE=$TemplateId";
	}

	return ( $DecodeTemplateRef, $Error );

}

#################### START sub header_decode() #############
sub header_decode {
	my ( $NetFlowPktRef, $OffSetRef ) = @_;
	my %NetFlowHeader = ();
	my $error         = undef;

	#
	# Extract Version
	#

	( $NetFlowHeader{VersionNum} ) = unpack( 'n', $$NetFlowPktRef );

	$$OffSetRef += 2;

	if ( $NetFlowHeader{VersionNum} == IPFIX ) {

		( @NetFlowHeader{qw{Length UnixSecs SequenceNum ObservationDomainId}} ) = unpack( "x$$OffSetRef nN3", $$NetFlowPktRef );

		$$OffSetRef += 2 + 4 * 3;

	} elsif ( $NetFlowHeader{VersionNum} == NetFlowv9 ) {

		( @NetFlowHeader{qw{Count SysUpTime UnixSecs SequenceNum SourceId}} ) = unpack( "x$$OffSetRef nN4", $$NetFlowPktRef );

		$$OffSetRef += 2 + 4 * 4;

	} elsif ( $NetFlowHeader{VersionNum} == NetFlowv8 ) {
	} elsif ( $NetFlowHeader{VersionNum} == NetFlowv5 ) {

		my $Sampling = undef;

		(   @NetFlowHeader{
				qw{Count SysUpTime UnixSecs UnixNsecs FlowSequenceNum
					EngineType EngineId}
			},
			$Sampling
			)
			= unpack(
			"x$$OffSetRef nN4C2n",
			$$NetFlowPktRef
			);

		$NetFlowHeader{SamplingMode}     = $Sampling >> 14;
		$NetFlowHeader{SamplingInterval} = $Sampling & 0x3FFF;

		$$OffSetRef += 2 * 1 + 4 * 4 + 1 * 2 + 2 * 1;

	}

	return ( \%NetFlowHeader, $error );

}
#################### END sub header_decode() ###############

#################### START sub flowset_decode() ############
sub flowset_decode {
	my ( $NetFlowPktRef, $OffSetRef ) = @_;
	my %FlowSetHeader = ();
	my @errors        = ();
	my $error         = undef;

	( $FlowSetHeader{SetId}, $FlowSetHeader{Length} ) = unpack( "x$$OffSetRef n2", $$NetFlowPktRef );

	$$OffSetRef += 2 * 2;

	return ( \%FlowSetHeader );

}
#################### END sub flowset_decode() ##############

#################### START sub template_decode() ###########
sub template_decode {
	my ( $NetFlowPktRef, $OffSetRef, $FlowSetHeaderRef, $VerNumRef ) = @_;
	my %Template = ();
	my $error    = undef;

	$Template{SetId} = $FlowSetHeaderRef->{SetId};

	#
	# decode data template for NetFlow v9 or IPFIX
	#

	if (   $FlowSetHeaderRef->{SetId} == NFWV9_DataTemplateSetId
		|| $FlowSetHeaderRef->{SetId} == IPFIX_DataTemplateSetId ) {

		( @Template{qw{TemplateId FieldCount}} ) = unpack( "x$$OffSetRef n2", $$NetFlowPktRef );

		$$OffSetRef += 2 * 2;

		#
		# decode option template for IPFIX
		#

	} elsif ( $FlowSetHeaderRef->{SetId} == IPFIX_OptionTemplateSetId ) {

		( @Template{qw{TemplateId FieldCount}} ) = unpack( "x$$OffSetRef n2", $$NetFlowPktRef );

		$$OffSetRef += 2 * 2;

		#
		# template withdraw check
		#

		if ( $Template{FieldCount} != 0 ) {

			( $Template{ScopeCount} ) = unpack( "x$$OffSetRef n", $$NetFlowPktRef );
			$$OffSetRef += 2 * 1;

		}

		#
		# decode option template for NetFlow v9
		#

	} elsif ( $FlowSetHeaderRef->{SetId} == NFWV9_OptionTemplateSetId ) {

		( @Template{qw{TemplateId OptionScopeLength OptionLength}} ) = unpack( "x$$OffSetRef n3", $$NetFlowPktRef );

		$$OffSetRef += 2 * 3;

		$Template{FieldCount} = int( ( $Template{OptionScopeLength} + $Template{OptionLength} ) / 4 );

		$Template{ScopeCount} = int( ( $Template{OptionScopeLength} ) / 4 );

	}

	return ( undef, 'ERROR: No fieldcount' ) if ( !defined( $Template{FieldCount} ) );

	for ( my $n = 0; $n < $Template{FieldCount}; $n++ ) {

		if ( $FlowSetHeaderRef->{SetId} <= IPFIX_OptionTemplateSetId ) {

			( @{$Template{Template}->[$n]}{qw{Id Length}} ) = unpack( "x$$OffSetRef n2", $$NetFlowPktRef );
			$$OffSetRef += 2 * 2;

			#
			# check enterprise number
			#

			if ( $$VerNumRef >= 10 ) {

				if ( $Template{Template}->[$n]->{Id} & 0x8000 ) {

					$Template{Template}->[$n]->{Id} -= 0x8000;

					( $Template{Template}->[$n]->{EnterpriseNum} ) = unpack( "x$$OffSetRef N", $$NetFlowPktRef );

          # We have a PEN add it to the Id.
					$Template{Template}->[$n]->{Id} =
            join('.', @{$Template{Template}->[$n]}{qw{EnterpriseNum Id}});

					$$OffSetRef += 4;

				}

			}

		}

	}

	return ( \%Template, $error );

}
#################### END sub template_decode() #############

#################### START sub flow_decode() ###############
sub flow_decode {
	my ( $NetFlowPktRef, $OffSetRef, $TemplateRef ) = @_;
	my %Flow   = ();
	my $error  = undef;
	my $Length = undef;

	if ( defined $TemplateRef->{TemplateId} ) {

		$Flow{SetId} = $TemplateRef->{TemplateId};

	} else {

		$error = 'ERROR: NOT FOUND TEMPLATE ID';

	}

	foreach my $ref ( @{ $TemplateRef->{Template} } ) {

		#
		# Variable Length Type
		#

		if ( $ref->{Length} == VariableLength ) {

			$Length = unpack( "x$$OffSetRef C", $$NetFlowPktRef );

			$$OffSetRef++;

			if ( $Length == 255 ) {

				$Length = unpack( "x$$OffSetRef n", $$NetFlowPktRef );

				$$OffSetRef += 2;

			}

			#
			# Fixed Length Type
			#

		} else {

			$Length = $ref->{Length};

		}

		#
		# One Template has multiple same Ids.
		#

		if ( defined $Flow{ $ref->{Id} } ) {

			my $Value = unpack( "x$$OffSetRef a$Length", $$NetFlowPktRef );

			$Flow{ $ref->{Id} } = [ $Flow{ $ref->{Id} } ] unless ref $Flow{ $ref->{Id} };

			push( @{ $Flow{ $ref->{Id} } }, $Value );

			#
			# Each Id is different than others.
			#

		} else {

			$Flow{ $ref->{Id} } = unpack( "x$$OffSetRef a$Length", $$NetFlowPktRef );

		}

		$$OffSetRef += $Length;

	}

	return ( \%Flow, $error );

}
#################### END sub flow_decode() #################

1;

__END__