abuseipdb_firewall.py

#!/usr/bin/python3
# coding=utf-8

from subprocess import run
import requests

abuseipdb_apikey = 'YOUR_API_KEY'
abuseipdb_url = 'https://api.abuseipdb.com/api/v2/blacklist'

set_name = 'abuseipdb_blacklist'

ipset_bin = '/sbin/ipset'
iptables_bin = '/sbin/iptables'

request_headers = {
	'Key': abuseipdb_apikey,
	'Accept': 'application/json'
}

run([ipset_bin, 'flush', set_name])
run([ipset_bin, 'create', set_name, 'iphash', '-exist'])

abuseipdb_response = requests.get(abuseipdb_url, headers=request_headers)

if abuseipdb_response:
	abuseipdb_blacklist = abuseipdb_response.json()
	
	try:
		for client_data in abuseipdb_blacklist['data']:
			run([ipset_bin, 'add', set_name, client_data['ipAddress'], '-exist'])
		#endfor
	except Exception as error:
		print('An error ocurred.')
		print(error)
	#endtry
#endif

#Only allow this line to run the first time (or when you restart the server). Then comment it to prevent issues.
run([iptables_bin, '-I', 'INPUT', '-m', 'set', '--match-set', set_name, 'src', '-j', 'DROP'])

abuseipdb_firewall_ipv6_support.py

#!/usr/bin/python3
# coding=utf-8

from subprocess import run
import requests
import ipaddress

abuseipdb_apikey = 'YOUR_API_KEY'
abuseipdb_url = 'https://api.abuseipdb.com/api/v2/blacklist'

set_for_ipv4 = 'abuseipdb_blacklist'
set_for_ipv6 = 'abuseipdb_blacklist_ipv6'

ipset_bin = '/sbin/ipset'
iptables_bin = '/sbin/iptables'
iptablesv6_bin = '/sbin/ip6tables'

request_headers = {
	'Key': abuseipdb_apikey,
	'Accept': 'application/json'
}

run([ipset_bin, 'flush', set_for_ipv4])
run([ipset_bin, 'flush', set_for_ipv6])

run([ipset_bin, 'create', set_for_ipv4, 'iphash', '-exist'])
run([ipset_bin, 'create', set_for_ipv6, 'iphash', '-exist', 'family', 'inet6'])

abuseipdb_response = requests.get(abuseipdb_url, headers=request_headers)

if abuseipdb_response:
	abuseipdb_blacklist = abuseipdb_response.json()
	
	try:
		for client_data in abuseipdb_blacklist['data']:
			try:
				ip_data = ipaddress.ip_address(client_data['ipAddress'])
				
				set_name_to_use = ''
				if( 4 == ip_data.version and ip_data.is_global ):
					set_name_to_use = set_for_ipv4
				elif( 6 == ip_data.version and ip_data.is_global ):
					set_name_to_use = set_for_ipv6
				#endif
					
				run([ipset_bin, 'add', set_name_to_use, client_data['ipAddress'], '-exist'])
				
			except ValueError:
				print('%s is not a valid IP address' % (client_data['ipAddress']))
			#endtry
		#endfor
	except Exception as error:
		print('An error ocurred.')
		print(error)
	#endtry
#endif

#Only allow this lines to run the first time (or when you restart the server). Then comment them to prevent issues.
run([iptables_bin, '-I', 'INPUT', '-m', 'set', '--match-set', set_for_ipv4, 'src', '-j', 'DROP'])
run([iptablesv6_bin, '-I', 'INPUT', '-m', 'set', '--match-set', set_for_ipv6, 'src', '-j', 'DROP'])

importabuseipdbblacklist.py

#!/usr/bin/python3
# coding=utf-8

# This file is meant to be added to the cron file and be executed at least once a day.
#
# The functions here defined are meant to be run also when the system starts. Please see loadabuseipdbonboot.py for more details.

from subprocess import run
import requests
import ipaddress

abuseipdb_apikey = 'YOUR_API_KEY'
abuseipdb_url = 'https://api.abuseipdb.com/api/v2/blacklist'

set_for_ipv4 = 'abuseipdb_blacklist'
set_for_ipv6 = 'abuseipdb_blacklist_ipv6'

ipset_bin = '/sbin/ipset'
iptables_bin = '/sbin/iptables'
iptablesv6_bin = '/sbin/ip6tables'


#Creates the needed ipsets. If they exist, fails silently.
def create_ip_sets():
	global set_for_ipv4, set_for_ipv6
	
	run([ipset_bin, 'create', set_for_ipv4, 'iphash', '-exist'])
	run([ipset_bin, 'create', set_for_ipv6, 'iphash', '-exist', 'family', 'inet6'])
#enddef


def import_abuseipdb_blacklist():
	global abuseipdb_apikey, abuseipdb_url, set_for_ipv4, set_for_ipv6
	global ipset_bin, iptables_bin, iptablesv6_bin

	request_headers = {
		'Key': abuseipdb_apikey,
		'Accept': 'application/json'
	}

	create_ip_sets()
	
	run([ipset_bin, 'flush', set_for_ipv4])
	run([ipset_bin, 'flush', set_for_ipv6])

	abuseipdb_response = requests.get(abuseipdb_url, headers=request_headers)

	if( abuseipdb_response):
		abuseipdb_blacklist = abuseipdb_response.json()
		
		try:
			for client_data in abuseipdb_blacklist['data']:
				try:
					ip_data = ipaddress.ip_address(client_data['ipAddress'])
					
					set_name_to_use = ''
					if( 4 == ip_data.version and ip_data.is_global ):
						set_name_to_use = set_for_ipv4
					elif( 6 == ip_data.version and ip_data.is_global ):
						set_name_to_use = set_for_ipv6
					#endif
						
					run([ipset_bin, 'add', set_name_to_use, client_data['ipAddress'], '-exist'])
					
				except ValueError:
					print('%s is not a valid IP address' % (client_data['ipAddress']))
				#endtry
			#endfor
		except Exception as error:
			print('An error ocurred.')
			print(error)
		#endtry
	#endif
#enddef

if( "__main__" == __name__ ):
	import_abuseipdb_blacklist()
#endif

loadabuseipdbonboot.py

#!/usr/bin/python3
# coding=utf-8

# This file is meant to be run only when the system starts up (or the very first time this is implemented on a system).
#
# This is because the iptables and ipsets are cleared whenver the system is restarted. Also, for the way this is implemented, the blacklist
# will be requested every time the machine is restarted or this script is run.
#
# If you need a much more "persistent" way to keep the list up, you can use the next commands:
#	ipset save [set_name] > /path/to/backup
#	ipset restore -exist < /path/to/backup
#
# With both commands you can easily backup and restore the sets you want.

from importabuseipdbblacklist import *

if( "__main__" == __name__ ):
	create_ip_sets()

	run([iptables_bin, '-I', 'INPUT', '-m', 'set', '--match-set', set_for_ipv4, 'src', '-j', 'DROP'])
	run([iptablesv6_bin, '-I', 'INPUT', '-m', 'set', '--match-set', set_for_ipv6, 'src', '-j', 'DROP'])

	import_abuseipdb_blacklist()
#endif