threatwall.py

#!/usr/bin/python2.7
# THREATWALL - block indicators of compromise (supported ones) using iptables from a git synchronized list
# This is currently alpha grade, use the -T option to set an "ACCEPT" target to see what type of traffic will be blocked before using it in production
# you need git,iptables and iptables-restore installed for this to work. 
# As it is now, it assumes the threat is external and it is running at the perimeter. this affects which direction the rules are applied.

import logging
import subprocess
import time
import os,sys,re,getopt

class IoC_Block:
	def __init__(self,ioc_repo,log_level=5,log_uid=True,
				log_prefix="THREATWALL blocked: ",target="DROP",ipblocks="/ioc/IPv4",domainblocks="/ioc/domain",urlblocks="/ioc/URL",
				inpolicy="ACCEPT",outpolicy="ACCEPT",forwardpolicy="ACCEPT",tmpath='/tmp/'):
		self.repo=os.path.abspath(ioc_repo)
		self.ip4file=ipblocks
		self.domainfile=domainblocks
		self.URLfile=urlblocks
		self.logprefix='"'+log_prefix.replace("'","")+'"'
		self.loglevel=str(log_level)
		self.target=target
		self.loguid=log_uid
		self.ready=False #didn't wanna use __new__()
		self.tmpath=tmpath		
		self.devnull=open("/dev/null","w")
		self.prelude='''#Rules generated by Threatwall
*filter
:INPUT '''+inpolicy+''' [0:0]
:FORWARD '''+forwardpolicy+''' [0:0]
:OUTPUT '''+outpolicy+''' [0:0]
:THREATWALL - [0:0]
'''
	
		if not os.path.exists(self.repo):
			print("Error,IOC repository not initialized.")
			logging.error("Error,IOC repository not initialized.")
			self.ready=False
			return
			subprocess.call(["/sbin/iptables","-X","THREATWALL"],shell=False,stdout=self.devnull)
		ret=subprocess.call(["/sbin/iptables","-S","THREATWALL"],shell=False,stdout=self.devnull)
		if ret == 1:
			ret=subprocess.call(["/sbin/iptables","-N","THREATWALL"],shell=False,stdout=self.devnull)
			if ret != 0:
				logging.error("Error creating a new chain")
		
		self.prelude+=' '.join(["-I","THREATWALL","-j",self.target]).strip()+"\n"
			
		if self.loguid:
			self.prelude+=' '.join(["-I","THREATWALL","-j","LOG","--log-prefix",self.logprefix,"--log-level",self.loglevel,"--log-uid"]).strip()+"\n"
		else:
			self.prelude+=' '.join(["-I","THREATWALL","-j","LOG","--log-prefix",self.logprefix,"--log-level",self.loglevel]).strip()+"\n"
			
		self.ready=True		
		
				
	def block_ip4(self):
		try:
			logging.info("Starting IPv4 blocks...")
			print("Starting IPv4 blocks...")
			ip4list=[]
			ipv4_pattern=re.compile("^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$")
			whitelist=set()
			with open(self.repo+"/whitelist.IPv4","a+") as wl4:
				for e in wl4.read().split("\n"):
					l=e.strip()
					if not None is e and len(e)>1 and not e[0] == "#":
						m=re.match(ipv4_pattern,l)
						if None is m:
							whitelist.add(l)
						else:
							print("Skipped IPv4 whitelist entry:"+l)
							logging.error("Skipped IPv4 whitelist entry:"+l)
							
			with open(self.repo+self.ip4file,"r") as ip4f:
				ip4list=ip4f.read().split("\n")
			iptlist_input=[]
			ret=subprocess.check_output(["/sbin/iptables","-S","INPUT"],shell=False)
			if not None is ret and len(ret)>0:
				iptlist_input=ret.split("\n")
			
			iptlist_output=[]
			ret=subprocess.check_output(["/sbin/iptables","-S","OUTPUT"],shell=False)
			if not None is ret and len(ret)>0:
				iptlist_output=ret.split("\n")
			blocklist=set()		
			for e in ip4list:
				if None is e or len(e)<2:
					continue
				#safety for the paranoid
				ip4=e.strip().replace(";","").replace("`","").replace(" ","") 
				m=re.match(ipv4_pattern,ip4)
				
				if None is m:
					print("Invalid entry in the IPv4 IOC list: "+e)
					logging.error("Invalid entry in the IPv4 IOC list: "+ip4)
					continue
				else:	
					#ntums
					skip_in=skip_out=skip_whitelist=False
					
					for ei in iptlist_input:
						if ip4 in ei:
							skip_in=True
							break
					for eo in iptlist_output:
						if ip4 in eo:
							skip_out=True
							break
							
					for ew in whitelist:
						if ip4 == ew:
							skip_whitelist=True
							break	
								
					if not skip_whitelist:	
						if skip_in:
							logging.debug("Skipping input block of "+ip4+" as a result of an exisiting block")
						else:
							blocklist.add(' '.join(["-A","INPUT","-s",ip4,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for inbound IPv4 IoC"']))
							
						if skip_out:
							logging.debug("Skipping output block of "+ip4+" as a result of an exisiting block")
						else:	
							blocklist.add(' '.join(["-A","OUTPUT","-d",ip4,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for outbound IPv4 IoC"']))
					else:
						print("Skipping "+ip4+" due to a whitelist entry.")		
						logging.info("Skipping "+ip4+" due to a whitelist entry.")
						
			tmp_path=self.tmpath+"/.threatwall.ipv4.iptables-save"
			with open(os.path.abspath(tmp_path),"a+") as ipts:
				ipts.write(self.prelude)
				for e in blocklist:
					ipts.write(e+"\n")
				ipts.write("COMMIT\n")
				
			ret=subprocess.call(["iptables-restore","--noflush","--verbose",tmp_path],shell=False) #calling iptables to apply this directly takes too long(as in hours)
			if ret != 0:
				print("Error applying IPv4 blocks via iptables-restore")
				logging.error("Error applying IPv4 blocks via iptables-restore")
			os.remove(tmp_path)
			print("Finished processing IPv4 blocks. "+str(len(blocklist))+" Blocks applied")
			logging.info("Finished processing IPv4 blocks. "+str(len(blocklist))+" Blocks applied")	
		except Exception as e:
			logging.exception("Error applying IPv4 blocks")	
			return 
			
	def block_domains(self): #based on the ipv4 block,with needed modifications.
		try:
			logging.info("Starting domain blocks...")
			print("Starting domain blocks...")
			domainlist=[]
			domain_pattern=re.compile("^[a-zA-Z0-9\.-]*$")
			with open(self.repo+"/"+self.domainfile,"r") as df:
				domainlist=df.read().split("\n")
			iptlist_input=[]
			whitelist=set()
			with open(self.repo+"/whitelist.domain","a+") as df:
				for e in df.read().split("\n"):
					l=e.strip()
					if not None is e and len(e)>1 and not e[0] == "#":
						m=re.match(domain_pattern,l)
						if None is m:
							whitelist.add(l)
						else:
							print("Skipped domain whitelist entry:"+l)
							logging.error("Skipped domain whitelist entry:"+l)
							
			ret=subprocess.check_output(["/sbin/iptables","-n","-L","INPUT"],shell=False)
			if not None is ret and len(ret)>0:
				iptlist_input=ret.split("\n")
			
			iptlist_output=[]
			ret=subprocess.check_output(["/sbin/iptables","-n","-L","OUTPUT"],shell=False)
			if not None is ret and len(ret)>0:
				iptlist_output=ret.split("\n")
				
			blocklist=set()	
			for e in domainlist:
				if None is e or len(e)<2:
					continue
				#safety for the paranoid
				domain=e.strip().replace(";","").replace("`","").replace(" ","")[:127] 
				m=re.match(domain_pattern,domain)
				if None is m:
					print("Invalid entry in the domain IOC list: "+domain)
					logging.error("Invalid entry in the domain IOC list: "+domain)
					continue
				#ntums
				skip_in=skip_out=skip_whitelist=False
				
				for ew in whitelist:
						if domain == ew:
							skip_whitelist=True
							break
							
				for e in iptlist_input:
					if domain.lower() in e.lower():
						skip_in=True
						break
						
				for e in iptlist_output:
					if domain.lower() in e.lower():
						skip_out=True
						break	
				hexdomain=''
				for l in domain.split("."):
					hexdomain+=format(len(l),'x').zfill(2)+l.encode('hex')
				hexdomain='"'+hexdomain[:128]+'"'	
				if not skip_whitelist:			
					if skip_in:
						logging.debug("Skipping input block of "+domain+" as a result of an exisiting block")
					else:
						blocklist.add(' '.join(["-A","INPUT","-p","tcp","--sport","53","-m","string","--algo","bm",'--hex-string',hexdomain,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for inbound TCP DNS IoC. Indicator:'+domain+'"']))
						blocklist.add(' '.join(["-A","INPUT","-p","udp","--sport","53","-m","string","--algo","bm",'--hex-string',hexdomain,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for inbound UDP DNS IoC. Indicator:'+domain+'"']))
					
					if skip_out:
						logging.debug("Skipping output block of "+domain+" as a result of an exisiting block")
					else:	
						blocklist.add(' '.join(["-A","OUTPUT","-p","tcp","--dport","53","-m","string","--algo","bm",'--hex-string',hexdomain,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for outbound TCP DNS IoC. Indicator:'+domain+'"']))
						blocklist.add(' '.join(["-A","OUTPUT","-p","udp","--dport","53","-m","string","--algo","bm",'--hex-string',hexdomain,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for outbound UDP DNS IoC. Indicator:'+domain+'"']))
				else:
					print("Skipping "+domain+" due to a whitelist entry.")		
					logging.info("Skipping "+domain+" due to a whitelist entry.")
								
			tmp_path= self.tmpath+"/.threatwall.domain.iptables-save"
			with open(os.path.abspath(tmp_path),"a+") as ipts:
				ipts.write(self.prelude)
				for e in blocklist:
					ipts.write(e+"\n")
				ipts.write("COMMIT\n")
			ret=subprocess.call(["iptables-restore","--noflush","--verbose",tmp_path],shell=False) #calling iptables to apply this directly takes too long(as in hours)
			if ret != 0:
				print("Error applying domain blocks via iptables-restore")
				logging.error("Error applying domain blocks via iptables-restore")

			os.remove(tmp_path)
			print("Finished processing domain blocks. "+str(len(blocklist))+" Blocks applied")
			logging.info("Finished processing domain blocks. "+str(len(blocklist))+" Blocks applied")			
		except Exception as e:
			logging.exception("Error applying domain blocks")	
			return 
			
	def block_urls(self):
		try:
			logging.info("Starting URL blocks...")
			print("Starting URL blocks...")
			URLlist=[]
			URL_pattern=re.compile("^[\s\w]+:\/\/([:\-_a-zA-Z0-9\.]+)(\/?.*)$")
			with open(self.repo+"/"+self.URLfile,"r") as df:
				URLlist=df.read().split("\n")
			iptlist_input=[]
			whitelist=set()
			with open(self.repo+"/whitelist.URL","a+") as df:
				for e in df.read().split("\n"):
					l=e.strip()
					if not None is e and len(e)>1 and not e[0] == "#":
						m=re.match(URL_pattern,l)
						if None is m:
							whitelist.add(l)
						else:
							print("Skipped URL whitelist entry:"+l)
							logging.error("Skipped URL whitelist entry:"+l)
							
			ret=subprocess.check_output(["/sbin/iptables","-n","-L","INPUT"],shell=False)
			if not None is ret and len(ret)>0:
				iptlist_input=ret.split("\n")
			
			iptlist_output=[]
			ret=subprocess.check_output(["/sbin/iptables","-n","-L","OUTPUT"],shell=False)
			if not None is ret and len(ret)>0:
				iptlist_output=ret.split("\n")
				
			blocklist=set()	
			URL_list_processed=set()
			
			for e in URLlist:
				if None is e or len(e)<2:
					continue
				#safety for the paranoid
				URLtmp=e.strip().replace(";","").replace("`","").replace(" ","") 
				m=re.match(URL_pattern,URLtmp)
				if None is m:
					print("Invalid entry in the URL IOC list: "+e)
					logging.error("Invalid entry in the URL IOC list: "+e)
					continue
				elif not None is m.group(1):
					URL_list_processed.add(m.group(1)[:127])
				elif not None is m.group(2):
					URL_list_processed.add(m.group(2)[:127])
				else:
					print("URL entry matched regex but no groups found:"+e)	
					logging.error("URL entry matched regex but no groups found:"+e)	
					continue
					
			for	URL in URL_list_processed:
				#ntums
				skip_in=skip_out=skip_whitelist=False
				
				for ew in whitelist:
						if URL == ew:
							skip_whitelist=True
							break
							
				for e in iptlist_input:
					if URL.lower() in e.lower():
						skip_in=True
						break
						
				for e in iptlist_output:
					if URL.lower() in e.lower():
						skip_out=True
						break	
				if not skip_whitelist:			
					if skip_in:
						logging.debug("Skipping input block of "+URL+" as a result of an exisiting block")
					else:
						blocklist.add(' '.join(["-A","INPUT","-p","tcp","--match","multiport","!","--sports","443,8443,6697","-m","string","--algo","bm","--string",URL,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for inbound TCP URL IoC"']))
						blocklist.add(' '.join(["-A","INPUT","-p","udp","--match","multiport","!","--sports","443,8443,6697","-m","string","--algo","bm","--string",URL,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for inbound UDP URL IoC"']))
					
					if skip_out:
						logging.debug("Skipping output block of "+URL+" as a result of an exisiting block")
					else:	
						blocklist.add(' '.join(["-A","OUTPUT","-p","tcp","--match","multiport","!","--dports","443,8443,6697","-m","string","--algo","bm","--string",URL,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for outbound TCP URL IoC"']))
						blocklist.add(' '.join(["-A","OUTPUT","-p","udp","--match","multiport","!","--dports","443,8443,6697","-m","string","--algo","bm","--string",URL,"-j","THREATWALL","-m","comment","--comment",'"Placed by threatwall for outbound UDP URL IoC"']))
				else:
					print("Skipping "+URL+" due to a whitelist entry.")		
					logging.info("Skipping "+URL+" due to a whitelist entry.")
								
			tmp_path= self.tmpath+"/.threatwall.URL.iptables-save"
			with open(os.path.abspath(tmp_path),"a+") as ipts:
				ipts.write(self.prelude)
				for e in blocklist:
					ipts.write(e+"\n")
				ipts.write("COMMIT\n")
			ret=subprocess.call(["iptables-restore","--noflush","--verbose",tmp_path],shell=False) #calling iptables to apply this directly takes too long(as in hours)
			if ret != 0:
				print("Error applying URL blocks via iptables-restore")
				logging.error("Error applying URL blocks via iptables-restore")

			os.remove(tmp_path)
			print("Finished processing URL blocks. "+str(len(blocklist))+" Blocks applied")
			logging.info("Finished processing URL blocks. "+str(len(blocklist))+" Blocks applied")			
		except Exception as e:
			logging.exception("Error applying URL blocks")	
			return 
			

		
class Fetch:
	def __init__(self,repository,ioc_path):
		self.repo=repository
		self.iocpath=os.path.abspath(ioc_path)
		if not os.path.exists(self.iocpath):
			try:
				self.iocpath=os.path.abspath(self.iocpath)
				os.makedirs(self.iocpath)
			except Exception.OSError as e:
				pass
			except Exception as e:
				logging.exception("Error while creating the IoC path at "+self.iocpath)	
				
	def clone(self):
		try:
			ret=subprocess.call(["git","clone",self.repo,os.getcwd()],shell=False)
			if ret != 0:
				logging.error("Error clonning IOC repository")
				print("Error clonning IOC repository")
				return False
			return True
		except Exception as e:
			logging.exception("Exception while clonning remote IOC repository")
			return False
						
	def sync(self):
		try:
			os.chdir(self.iocpath)
			if not os.path.exists(self.iocpath+"/.git"):
				print ("Cloning IOC repository for the first time.")
				logging.info("Cloning IOC repository for the first time.")
				
				if self.clone() != True:
					return
			
			ret=subprocess.call(["git","pull"] ,shell=False)
			if ret != 0:
				print("Error pulling contents of remote repository.")
				logging.error("Error pulling contents of remote repository")
			else:
				logging.info("Synchronized with remote repository: "+self.repo)
					
		except Exception as e:
			logging.exception("Error while syncing  git repository for indicators")
			
def usage():
	print("Usage: python2 ",os.path.basename(__file__)," [-vhBSlTtcrfoi] ",('''
		-i	<DROP|ACCEPT|REJECT,..> INPUT chain policy
		-o  <DROP|ACCEPT|REJECT,..> OUTPUT chain policy
		-f  <DROP|ACCEPT|REJECT,..> FORWARD chain policy
		-r  <https://url/> remote git repository that stores the indicators
		-c  <fspath> Filesystem path used to clone the remote repository to
		-t  </temporarypath/> Filesystem path used to store temporary files
		-T  <DROP|ACCEPT|REJECT,..> iptables chain 'policy' target for the THREATWALL chain
		-l  <path> Filesystem path used to store logfiles
		-S  <secs> Interval in seconds between syncing of the remote git repository
		-B  Run in background (Foreground is default)
		-v  Print this applications current version string
		-h  Display this usage instruction.'''))

def main():
	args={}
	if os.geteuid() != 0:
		print("This program needs to run as root.")
		return 1
	else:	
		try:
			opts, args = getopt.getopt(sys.argv[1:], "i:o:f:r:c:t:T:l:Bvh")
		except getopt.GetoptError as err:
			print (str(err))
			self.usage()		
			return 1
		args={'INPUT-POLICY':'ACCEPT',
			'OUTPUT-POLICY':'ACCEPT',
			'FORWARD-POLICY':'ACCEPT',
			'logpath':'/var/log/threatwall/',
			'gitrepo':'https://github.com/MainframeSkuzzy/threatwall',
			'localclone':'/etc/threatwall/synced',
			'tmpath':'/tmp',
			'target':'DROP', #be sure to never set this to 'THREATWALL' as it would be sending rules to itself
			'mode':'foreground',
			'interval':1800
				}	
		for opt, arg in opts:
			if opt== '-h':
				usage()
				return 0
			if opt== '-v':
				print("Version: 0.1a")
				return 0
			if opt=='-i':
				args['INPUT-POLICY']=arg
			if opt=='-o':
				args['OUTPUT-POLICY']=arg
			if opt=='-f':
				args['FORWARD-POLICY']=arg
			if opt=='-l':
				args['logpath']=os.path.abspath(arg)
			if opt=='-r':
				args['gitrepo']=arg
			if opt=='-c':
				args['localclone']=os.path.abspath(arg)
			if opt=='-t':
				args['tmpath']=os.path.abspath(arg)
			if opt=='-T':
				args['target']=arg
			if opt=='-S':
				args['interval']=int(arg,10)
			if opt=='-B':
				args['mode']="background"
					
	if not os.path.exists(args['logpath']):
		os.makedirs(args['logpath'],0754)
	if not os.path.exists(args['localclone']):
		os.makedirs(args['localclone'],0754)
	if not os.path.exists(args['tmpath']):
		os.makedirs(args['tmpath'],0754)
								
	logging.basicConfig(filename=args['logpath']+"/threatwall.log",level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
	
	if args['mode'] == 'background' and os.fork():
		sys.exit()
		
	feed=Fetch(args['gitrepo'],args['localclone'])
	ioc=IoC_Block(args['localclone'],target=args['target'],inpolicy=args['INPUT-POLICY'],outpolicy=args['OUTPUT-POLICY'],forwardpolicy=args['FORWARD-POLICY'],tmpath=args['tmpath'])
	feed.sync() #the above line needs to happen before this.

	if ioc.ready:
		while True:
			ioc.block_ip4()
			ioc.block_domains()
			ioc.block_urls() 
			logging.info("Applied all blocks")
			time.sleep(args['interval']) 
			feed.sync()
			logging.info("Synchronized indicator list.")
			
if __name__ == "__main__":
	main()