monitor

#!/usr/bin/perl

# -----------------------------------------------------------------------------
#
# RBL Updater Suite - a suite to update realtime blocklists for rspamd
# Copyright (C) 2022 - John Bradley (userjack6880)
#
# monitor
#   monitors mail log for rejected emails and reports IP addresses to the
#   system
#
# Available at: https://github.com/userjack6880/rbl_updater
#
# -----------------------------------------------------------------------------
#
# This file is part of the RBL Updater Suite for use with rspamd
#
# The RBL Updater Suite 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 3 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, see <https://www.gnu.org/licenses/>.
#
# -----------------------------------------------------------------------------

use strict;
use warnings;

use Getopt::Std;
use File::Tail;
use File::Basename;
use JSON;
use LWP::UserAgent;
use LWP::Protocol::https;
use DBI;

# version number
my $version = '1 Feature Complete';

# basic settings --------------------------------------------------------------
my %opts = ();
getopts('c:vi:sh', \%opts);

my $verbose     = ($opts{'v'} || 0);
my $import_file = ($opts{'i'} || '');
my $tail_start = 0;
if (defined $opts{'s'}) {
    $tail_start = '-'.$opts{'s'};
} 
my $conffile = ($opts{'c'} || 'config.conf');

# usage sub -------------------------------------------------------------------
sub usage {
	print "\n".
	      "\tUsage:\n".
	      "\t\t./monitor [opts]\n".
	      "\n".
	      "\tThis script monitors the mail log defined in the config file in real\n".
	      "\ttime and updates the database accordingly.\n".
	      "\n".
	      "\tOptions:\n".
          "\t\t-c [confpath]\tLoad Config File\n".
	      "\t\t-v\t\tVerbose Mode\n".
          "\t\t-i [logpath]\tImport Log\n".
          "\t\t-s\t\tStart Tail at Beginning\n".
	      "\n";
}

if (defined $opts{'h'}) {
    usage();
    exit;
}

# load config file ------------------------------------------------------------
our ($asnlist,$iplist,$dbname,$dbhost,$dbport,$dbuser,$dbpass,$maillog,$log);
my @ipqueue;

if (substr($conffile,0,1) ne '/' and substr($conffile,0,1) ne '.') { $conffile = "./$conffile"; }

if (-e File::Basename::dirname($0)."/$conffile") { $conffile = File::Basename::dirname($0)."/$conffile"; }
unless (-e $conffile) { usage(); die "Error! Could not read $conffile\n"; }

my $conftest = do $conffile;
die "$conffile could not be parsed: $@" if $@;
die "could not do $conffile: $!" if !defined $conftest;

# declare "global" variables --------------------------------------------------
my $dbh;

# BGP View Queries ------------------------------------------------------------
sub query_ip {
	my $ip = shift;
	my $result;

	my $ua = LWP::UserAgent->new(send_te => 0);
	$ua->agent("Mozilla/8.0"); # because cloudflare is a bitch
	my $req = HTTP::Request->new(
	            GET => "https://api.bgpview.io/ip/$ip",
	            [
	              'Content-Type' => "application/json"
	            ]
	          );
	my $res = $ua->request($req);
	# return error if JSON isn't returned
	if (substr($res->decoded_content,0,1) eq '<') {
        logme("bgpview returned nothing");
		$result->{error} = 1;
		return $result;
	}
	my $data = decode_json($res->decoded_content);
	$result->{prefix}   = $data->{data}{rir_allocation}{prefix};
	$result->{prefix}   = $data->{data}{prefixes}[0]{prefix} if not defined($result->{prefix});
	$result->{asn}      = $data->{data}{prefixes}[0]{asn}{asn};
    my $asn;
    if (defined($result->{asn})) { 
        $asn = $result->{asn}; 
    }
    else { 
        $asn = 'Unknown'; 
    }
	$result->{provider} = $data->{data}{prefixes}[0]{asn}{description};
	$result->{provider} = $data->{data}{prefixes}[0]{description} if not defined($result->{provider});
	$result->{provider} = 'Unknown' if not defined($result->{provider});

    logme("bgpview results\tprefix: ".$result->{prefix}."\tasn: $asn\tprovider: ".$result->{provider});
	return $result;
}

sub query_asn {
	my $asn = shift;
	my @result;

	return @result if not defined($asn);

	my $ua = LWP::UserAgent->new(send_te => 0);
	$ua->agent("Mozilla/8.0"); # because cloudflare is a bitch
	my $req = HTTP::Request->new(
	            GET => "https://api.bgpview.io/asn/$asn/prefixes",
	            [
	              'Content-Type' => "application/json"
	            ]
	          );
	my $res = $ua->request($req);
	if (substr($res->decoded_content,0,1) eq '<') {
        logme("bgpview returned nothing");
		$result[0] = 'error';
		return @result;
	}
	my $data = decode_json($res->decoded_content);
	my $itr = 0;
	foreach my $entry (@{$data->{data}{ipv4_prefixes}}) {
		$result[$itr] = $entry->{prefix};
        logme("prefix ".$result[$itr]." belongs to AS$asn");
		$itr++;
	}

	return @result;
}

# DB Related Subroutines ------------------------------------------------------
sub db_keepalive {
	if(!$dbh->ping) {
		$dbh = DBI->connect_cached(
             "DBI:mysql:database=$dbname;host=$dbhost;port=$dbport",
             $dbuser,
             $dbpass
           ) or die "Could not connect to database!\n";
	}
}

sub add_asn {
	my $asn = shift;
	my $provider = shift;
	db_keepalive();
	my $sth;

	# the most severe of penalties - absolutely will have collateral damage
	# - initially based on percentage of ranges that get permabanned, only add to spam
	#   scores
	# - when threshold is reached (>50%), ASN gets banned for increasing amounts of time

	# when the recusive call returns, we can continue;

	print "FLAGGED: AS$asn - $provider\n" if $verbose;
    logme("FLAGGED: AS$asn - $provider");

	# check if the ASN has been recorded
	# if it doesn't exist, put it in the database
	if ($dbh->selectrow_array("SELECT EXISTS(SELECT * FROM asn_blocklist WHERE asn = '$asn');") < 1) {

		# we need to record the ASN
		$sth = $dbh->prepare(q{
		  INSERT INTO asn_blocklist (asn, provider, blocked_ranges, total_ranges, infractions, infractions_type, permaban)
		  VALUES (?,?,1,?,0,1,0);
		});

		# asn query now...
		my @ranges = query_asn($asn);
		if (!@ranges) {
			print "no ASN info! skipping\n" if $verbose;
            logme("no ASN info! skipping");
			return;
		}
		if ($ranges[0] eq 'error') {
			print "JSON error. Skipping\n";
            logme("JSON error. Skipping");
			return;
		}
		else {
			print "AS$asn has ".scalar(@ranges)." ranges\n" if $verbose;
            logme("AS$asn has ".scalar(@ranges)." ranges");

			$sth->execute($asn,$provider,scalar(@ranges));
		}
	}
	else {
		# instead of recording the ASN, we now update the ASN record

		$sth = $dbh->prepare(q{
		  UPDATE asn_blocklist
		  SET blocked_ranges = blocked_ranges + 1
		  WHERE asn = ?;
		});
		$sth->execute($asn);

		# next, we need to count how many banned ranges are in this ASN, then determine action
		my ($banned,$total,$infractions) = $dbh->selectrow_array("SELECT blocked_ranges,total_ranges,infractions FROM asn_blocklist WHERE asn = '$asn';");
		print "AS$asn has $total ranges. $banned ranges are banned. AS$asn has $infractions infractions\n" if $verbose;
        logme("AS$asn has $total ranges. $banned ranges are banned. AS$asn has $infractions infractions");

		# because this will never be the first time this happens, we can safey process actions here...
		# check if 50% has been exceeded
		my $percent = sprintf("%.0f",($banned/$total)*100);
		print "$percent% banned\n" if $verbose;
        logme("$percent% banned");

		if ($percent > 50) {
			print "threshold exceeded!\n" if $verbose;
            logme("threshold exceeded!");

			my @ranges = query_asn($asn);
			if (!@ranges) {
				print "no ASN info!\n" if $verbose;
                logme("no ASN info! skipping");
				return;
			}
			if ($ranges[0] eq 'error') {
				print "JSON error\n" if $verbose;
                logme("JSON error. Skipping");
				return;
			}
			# we need to increase the infraction count, or permaban if infraction count is exceeded
			if ($infractions == 0) {
				print "AS$asn: first infraction, 1 week ban\n" if $verbose;
                logme("AS$asn: first infraction, 1 week ban");
				$sth = $dbh->prepare(q{
				  UPDATE asn_blocklist
				  SET infractions = infractions +1
				  WHERE asn = ?
				  AND permaban = 0;
				});
				$sth->execute($asn);
				$sth = $dbh->prepare(q{
				  INSERT INTO ipnet_blocklist (ip4_net, asn, provider, infractions, infractions_type, ban_expiration, permaban)
				  VALUES (?,?,?,0,1,DATE_ADD(NOW(), INTERVAL 1 WEEK),0)
				  ON DUPLICATE KEY UPDATE ban_expiration = CASE
				    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 1 WEEK)
				    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 1 WEEK)
				  END;
				});
			}
			elsif ($infractions == 1) {
				print "AS$asn: second infraction, 1 week ban\n" if $verbose;
                logme("AS$asn: second infraction, 1 week ban");
				$sth = $dbh->prepare(q{
				  UPDATE asn_blocklist
				  SET infractions = infractions +1
				  WHERE asn = ?
				  AND permaban = 0;
				});
				$sth->execute($asn);
				$sth = $dbh->prepare(q{
				  INSERT INTO ipnet_blocklist (ip4_net, asn, provider, infractions, infractions_type, ban_expiration, permaban)
				  VALUES (?,?,?,0,1,DATE_ADD(NOW(), INTERVAL 1 MONTH),0)
				  ON DUPLICATE KEY UPDATE ban_expiration = CASE
				    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 1 MONTH)
				    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 1 MONTH)
				  END;
				});
			}
			else {
				print "AS$asn: third infraction, perma ban\n" if $verbose;
                logme("AS$asn: third infraction, permaban");
				$sth = $dbh->prepare(q{
				  UPDATE asn_blocklist
					SET permaban = 1
				  WHERE asn = ?
				  AND permaban = 0;
				});
				$sth->execute($asn);
				$sth = $dbh->prepare(q{
				  INSERT INTO ipnet_blocklist (ip4_net, asn, provider, infractions, infractions_type, ban_expiration, permaban)
				  VALUES (?,?,?,0,1,NOW(),1)
				  ON DUPLICATE KEY UPDATE permaban = 1;
				});
			}
			foreach my $range (@ranges) {
				print "\tBan added for $range\n" if $verbose;
                logme("Ban added for $range");
				$sth->execute($range,$asn,$provider);
			}
		}
	}
}

sub add_net {
	my $range = shift;
	db_keepalive();
	my $prefix = $range->{prefix};
	my $asn = $range->{asn};
	my $provider = $range->{provider};

	# at this point, we've gotta be cautious, as this may create collateral damage
	print "FLAGGED: $prefix - $provider\n" if $verbose;
    logme("FLAGGED: $prefix - $provider");
	my $sth = $dbh->prepare(q{
	  INSERT INTO ipnet_blocklist (ip4_net, asn, provider, infractions, infractions_type, ban_expiration, permaban)
		VALUES (?,?,?,1,1,NOW(),0)
		ON DUPLICATE KEY UPDATE infractions = infractions + 1,
	                          ban_expiration = CASE
	                            WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 1 HOUR)
	                            WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 1 HOUR)
	                          END;
	});
	$sth->execute($prefix,$asn,$provider);

	# increment any time-based banned based on infractions...
	my $infractions = $dbh->selectrow_array("SELECT infractions FROM ipnet_blocklist WHERE ip4_net = '$prefix';");

	print "\t$infractions range infractions\n" if $verbose;
    logme("$infractions range infractions");

	if ($infractions > 24) {
		print "$prefix is now permanently banned!\n" if $verbose;
        logme("$prefix is now permanently banned!");
		# add prefix to permaban, note infraction in asn
		$sth = $dbh->prepare(q{
		  UPDATE ipnet_blocklist
		  SET permaban = 1
		  WHERE ip4_net = ?
		  AND permaban = 0;
		});
		$sth->execute($prefix);

		add_asn($asn,$provider) if $sth->rows >= 1;
	}
	else {
		if ($infractions == 3) {
			$sth = $dbh->prepare(q{
			  UPDATE ipnet_blocklist
			  SET ban_expiration = CASE
			    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 1 DAY)
			    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 1 DAY)
			  END
			  where ip4_net = ?;
			});
			print "$prefix has recieved a 1 day ban\n" if $verbose;
            logme("$prefix has recieved a 1 day ban");
		}
		if ($infractions > 3 && $infractions < 25) {
			$sth = $dbh->prepare(q{
			  UPDATE ipnet_blocklist
			  SET ban_expiration = CASE
			    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 1 WEEK)
			    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 1 WEEK)
			  END
			  where ip4_net = ?;
			});
			print "$prefix has recieved a 1 week ban\n" if $verbose;
            logme("$prefix has recieved a 1 week ban");
		}

		$sth->execute($prefix) if $infractions > 3;
	}
}

sub add_ip {
	my $ip = shift;
	db_keepalive();

	print localtime().": " if $verbose;

	# first check if we can even process the ip, if not, throw it on the queue
	my $range = query_ip($ip);
	if (defined $range->{error}) {
		push(@ipqueue, $ip);
		print "JSON error. Saving $ip in queue!\tQueue has ".scalar(@ipqueue)." IPs\n" if $verbose;
        logme("JSON error. Saving $ip in queue!\tQueue has ".scalar(@ipqueue)." IPs");
		return;
	}

	# next we should process through queued IPs
	my $queued = pop(@ipqueue);

    add_ip($queued) if defined($queued);

	# now we process the IP we just got
	print "FLAGGED: $ip - ".$range->{provider}."\n" if $verbose;

	# simple increment, create if it doesn't exist
	my $sth = $dbh->prepare(q{
	  INSERT INTO ip_blocklist (ip4, ip4_net, asn, provider, infractions, infractions_type, ban_expiration, permaban)
	  VALUES (INET_ATON(?),?,?,?,1,1,NOW(),0)
	  ON DUPLICATE KEY UPDATE infractions = infractions + 1;
	});
	$sth->execute($ip,$range->{prefix},$range->{asn},$range->{provider});

	# increment time-based bans, set permabans as needed
	my $infractions = $dbh->selectrow_array("SELECT infractions FROM ip_blocklist WHERE ip4 = INET_ATON('$ip');");

	print "\t$infractions infractions\n" if $verbose;
    logme("FLAGGED: $ip - ".$range->{provider}."\t$infractions infractions");

	# check to see if there are more than 5 address in the net that are currently banned... if so, then more severely punish the prefix
	# does not penalize for permabanned IPs
	if ($dbh->selectrow_array("SELECT COUNT(*) FROM ip_blocklist WHERE ip4_net = '$range->{prefix}' AND ban_expiration > NOW();") > 5) {
		print "\tUnusually high infraction from net $range->{prefix}. Punishing\n" if $verbose;
        logme("Unusually high infraction from net ".$range->{prefix}.". Punishing");

		add_net($range);
	}

	if ($infractions > 3) {
		print "$ip is now permanently banned!\n" if $verbose;
        logme("$ip is now permanently banned!");
		# adds ip to permaban, add infraction to the IP4net
		$sth = $dbh->prepare(q{
		  UPDATE ip_blocklist 
		  SET permaban = 1
		  WHERE ip4 = INET_ATON(?)
			AND permaban = 0;
		});
		$sth->execute($ip);

		add_net($range) if $sth->rows >= 1;
	}
	else {
		if ($infractions == 1) {
			$sth = $dbh->prepare(q{
			  UPDATE ip_blocklist 
			  SET ban_expiration = CASE
			    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 1 HOUR)
			    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 1 HOUR)
			  END
			  WHERE ip4 = INET_ATON(?);
			});
			print "$ip has recieved a 1hr ban\n" if $verbose;
            logme("$ip has recieved a 1hr ban!");
		}

		if ($infractions == 2) {
			$sth = $dbh->prepare(q{
			  UPDATE ip_blocklist 
			  SET ban_expiration = CASE
			    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 6 HOUR)
			    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 6 HOUR)
			  END
			  WHERE ip4 = INET_ATON(?);
			});
			print "$ip has recieved a 6hr ban\n" if $verbose;
            logme("$ip has recieved a 6hr ban!");
		}

		if ($infractions == 3) {
			$sth = $dbh->prepare(q{
			  UPDATE ip_blocklist 
			  SET ban_expiration = CASE
			    WHEN ban_expiration < NOW() THEN DATE_ADD(NOW(), INTERVAL 12 HOUR)
			    WHEN ban_expiration >= NOW() THEN DATE_ADD(ban_expiration, INTERVAL 12 HOUR)
			  END
			  WHERE ip4 = INET_ATON(?);
			});
			print "$ip has recieved a 12hr ban\n" if $verbose;
            logme("$ip has recieved a 12hr ban!");
		}

		$sth->execute($ip);
	}

}

# Logging Mechanism -----------------------------------------------------------

sub logme {
    my $logmsg = shift;
    open (my $fh, '>>', $log) or die "Could not open log!\n";
    print $fh localtime()."\t$logmsg\n";
    close $fh;
}

# Process Lines ---------------------------------------------------------------

sub proc_line {
    my $line = shift;
    # rejects by postfix are autoinfractions
    if ($line =~ /^.+NOQUEUE: reject:.+/) {
		if ($line =~ m/^.+\[(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}).+/) {
            logme("matched on NOQUEUE - ($line)");
			add_ip($1);
		}
	}

	# rejects by rspamd might just be greylist bounces
	if ($line =~ /^.+milter-reject.+/) {
		# this will only penalize IPs that exit on the current blocklist or legit spam
		if ($line =~ /^.+BLOCKLIST.+/ || 
		    $line =~ /^.+spam.+/ ||
		    $line =~ /^.+Spam.+/ ) {
			if ($line =~ m/^.+\[(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}).+/) {
                logme("matched on milter-reject - ($line)");
				add_ip($1);
			}
		}
	}
}

# Begin Main Code -------------------------------------------------------------


print "\nRBL Updater Suite v.$version\n".
      "\tMonitor\n\n".
      "Monitoring $maillog\n" if $verbose;
logme("starting rbl updater suite v.$version monitor - monitoring $maillog");

# Open DB connection

logme("connecting to $dbname\@$dbhost:$dbport");
$dbh = DBI->connect(
            "DBI:mysql:database=$dbname;host=$dbhost;port=$dbport",
            $dbuser,
            $dbpass
          ) or die "Could not connect to database!\n";
logme("success!");

# optionally import logs and quit;
if ($import_file ne '') {
    logme("importing $import_file");
    open(my $file, '<', $import_file) or die "could not open log";
    while (my $line = <$file>) {
        proc_line($line);
    }
    logme("import done!");
}

# monitor log file and act when proper log arrives
else {
    my $file=File::Tail->new(name=>$maillog,tail=>$tail_start,reset_tail=>$tail_start);
    while (defined(my $line=$file->read)) {
    	proc_line($line);
    }
}

$dbh->disconnect;