Perl script for processing+merging multiple iptables/UFW logs

fwblocks.pl

#!/usr/bin/perl
use strict;
use warnings;
use autodie;
use DBI;
use utf8;
use Scalar::Util qw(looks_like_number);
use Data::Validate::IP qw( is_public_ip is_linklocal_ip is_loopback_ip is_private_ip );
use Geo::IP;
use File::Tail;
use Shell::Perl::Dumper;
use Tie::IxHash;
use Time::Piece;
use DateTime::Format::Strptime;
use Time::Local;
use Parse::Syslog;
use File::MergeSort;
use Data::Dumper;
use Hash::Ordered;
no warnings 'uninitialized';
no warnings 'numeric';

local $Data::Dumper::Terse = 0;
my @logs = qw(/var/log/ufw.log /var/log/router.log);
my $fmt = "%b %d %H:%M:%S";
my $gi = Geo::IP->open("/usr/share/GeoIP/GeoIPCity.dat", GEOIP_STANDARD);
my $db = DBI->connect("dbi:CSV:", undef, undef, {
		f_ext => ".csv/r",
		f_dir => "/etc",
		f_dir_search => [],
		csv_eol          => "\n",
		csv_sep_char     => ",",
		csv_quote_char   => '',
		csv_escape_char  => '',
		f_encoding => "utf8",
		RaiseError => 1,
		csv_tables       => {
			protocols => {
			sep_char    => ",",
			quote_char  => undef,
			escape_char => undef,
			file        => "/etc/protocols.csv",
			col_names   => [qw( number name )],
		},
},
	}) or die "Cannot connect: $DBI::errstr";
$db->{csv_allow_loose_quotes}  = 1;
$db->{csv_allow_loose_escapes} = 1;
$db->{csv_allow_whitespace}    = 1;

sub timestamp {
	return Time::Piece->strptime($_[0], "%Y%m%d%H%M%S");
}

sub epoch {
	my $line = shift;
    my $timestamp = $line =~ /^([A-Z][a-z][a-z]\s{1,2}\d{1,2}\s\d{2}[:]\d{2}[:]\d{2}).*/;
	my $parser = DateTime::Format::Strptime->new(
    	pattern => $fmt, 
	);
	my $dt = $parser->parse_datetime($1);
	#my $strp = DateTime::Format::Strptime->new( pattern => $fmt );
	my $epoch = $dt->epoch;
    return $epoch;
  }


sub numeric_date {
	my $line = defined($_[0]) ? $_[0] : shift;
	$line =~ /^([A-Z][a-z][a-z])\s{1,2}(\d{1,2})\s(\d{2}[:]\d{2}[:]\d{2}).*/;
	my %mon2num = qw( Jan 01 Feb 02 Mar 03 Apr 04 May 05 Jun 06 Jul 07 Aug 08 Sep 09 Oct 10 Nov 11 Dec 12 );
	my $year = 1900 + (localtime)[5];
	my $month = $mon2num{$1};
	my $day = sprintf("%02d", $2);
	my $time = $3;
	$time =~ tr/://d;
	my $date = sprintf("%d", "$year$month$day$time");
	$date += 0;
	return $date;
}

sub fwlogs {
	my @linez;
	my %opts = ( skip_empty_lines => 1 );
	my $ms = File::MergeSort->new( \@logs, \&numeric_date, \%opts);
	# for my $file (@logs) {
	# open(my $f, '<:encoding(UTF-8)', $file) or die "Could not open file '$file' $!";
	# while (my $line = <$f>) {
	while ( my $line = $ms->next_line() ) {
			my $timestamp = numeric_date($line);
			push @linez, { $timestamp => $line };
	}
	$ms->dump("/tmp/fwblocks.log");
	# } }
	#sort(keys(@linez))
	my @sorted = map { $_ } sort { $a <=> $b} (@linez);
	return @sorted;
	# return @lines;
}

sub geoip {
	my $ip = (defined $_[0] and length $_[0] > 0 and is_public_ip($_[0])) ? $_[0] : $_[1];
	my $geoip = $gi->record_by_addr($ip);
	return defined $geoip ? $geoip->city . " " . $geoip->region . " " . 	$geoip->region_name . " " . $geoip->country_name : undef;
}

sub proto {
	if (looks_like_number($_[0])) {
		my $protocol = $db->prepare("SELECT * from protocols WHERE number = " . $_[0]);
		$protocol->execute() || die "can't fetch all rows";
		while (my $row = $protocol->fetchrow_hashref) {
			$protocol->finish;
			return $row->{name};
		}
	} else {
		return undef;
	}
}

sub parse {
	$_[0] =~ m/.*IN=(\S*).*OUT=(\S*).*SRC=(\S*).*DST=(\S*).*PROTO=(\S*)(?: SPT=(\S*) DPT=(\S*))?.*/;
	my %ipblock = (
			'input' => $1,
			'output' => $2,
			'srcip' => $3,
			'dstip' => $4,
			'proto' => proto($5),
			'srcport' => $6,
			'dstport' => $7,
			'location' => geoip($3, $4),
			'timestamp' => timestamp($_[1]),
	);

	delete @ipblock{ grep { not defined $ipblock{$_} } keys %ipblock };
	delete @ipblock{ grep { length $ipblock{$_} eq 0 } keys %ipblock };;
	#tie %ipblock, "Tie::IxHash";
	#tie %ipblock, "Hash::Ordered", @ipblock;
	#my %ipb = Hash::Ordered->new(@ipblock);
	return %ipblock;
} 

sub trim { my $s = shift; $s =~ s/^\s+|\s+$//g; return $s };

sub main {
	my @lines = fwlogs();
	for my $line (@lines) {
		while( my( $key, $val ) = each %{$line} ){
			if ($val =~ m/.*(UFW BLOCK]\s|rtac88u kernel\: DROP )(.*)/) {
				my %newline = parse($2, $key);
				while( my( $k, $v ) = each %newline ){
					printf("%-6s: %-10s\t", trim($k), trim($v)) 
				}
			print "\n";
			} else {
				next;
			}
		}
	
	}
}

main();
$db->disconnect;

See lines 24 and 148 for path and regex customization. Here's the required /etc/protocols.csv:

number,name
0,IP
1,ICMP
2,IGMP
3,GGP
4,IPENCAP
5,ST
6,TCP
8,EGP
9,IGP
12,PUP
17,UDP
20,HMP
22,XNSIDP
27,RDP
29,ISOTP4
33,DCCP
36,XTP
37,DDP
38,IDPRCMTP
41,IPv6
43,IPv6Route
44,IPv6Frag
45,IDRP
46,RSVP
47,GRE
50,IPSECESP
51,IPSECAH
57,SKIP
58,IPv6ICMP
59,IPv6NoNxt
60,IPv6Opts
73,RSPF
81,VMTP
88,EIGRP
89,OSPFIGP
93,AX25
94,IPIP
97,ETHERIP
98,ENCAP
103,PIM
108,IPCOMP
112,VRRP
115,L2TP
124,ISIS
132,SCTP
133,FC
135,MobilityHeader
136,UDPLite
137,MPLSinIP
139,HIP
140,Shim6
141,WESP
142,ROHC