ZoneScanner.py

#!/usr/bin/python3 -O
"""
 File: 			ZoneScanner.py
 Author:	 	Leon McClatchey
 Date:	 		2025-11-04
 Description:	Scans Named (dns) query logs for signs of Bot Scanners
				Then blocks those discovered IP addresses
				This script is designed to be run on the local host,
				but it does pull information from the ansible yaml configuration files
				
"""
import argparse
import os, sys, subprocess
import socket
import logging
import pathlib
import re
# Initialize Global Strings
Config = {}

VAULT_FILE = "secrets.yml"
LOG_PATH = pathlib.Path("/var/lib/named/log/queries.log")
BLOCK_CMD_TEMPLATE = ["firewall-cmd", "--permanent", "--add-rich-rule=rule family='ipv4' source address='{}' reject"]
RELOAD_CMD = ["firewall-cmd", "--reload"]

# === Suspicious DNS Query Patterns ===
SUSPICIOUS_PATTERNS = [
    r"AXFR",             # Zone transfer attempts
    r"ANY",              # Amplification probes
    r"\.env",            # Misguided bot scans
    r"localhost\.localdomain",  # Misconfigured clients
]
# Function to check the system Path => Common
def CheckSysPath(target):
	npath = os.path.normcase(os.path.abspath(target))
	for p in sys.path:
	  nsys = os.path.normcase(os.path.abspath(p))
	  if npath == nsys: return True
	return False
# Function to obtain host information from ansible data files
def ConfigureHosts(parms):
	log = parms['log']['path']
	host = parms['host']
	key = parms['key']
	yml = os.path.join(parms['inventory'],"hosts.yml")
	if not os.path.isfile(yml):
		msg = f"{yml} does not point to a regular File, Aborting!"
		FileTools.WriteLog(msg,log,logging.CRITICAL) if log is not None else print(msg)
		sys.exit(5)
	Ydata = AnsibleTools.LoadYaml(yml,log)
	details = AnsibleTools.find_host_any_distro(Ydata,host)
	port = 22 if 'port' not in Ydata['all']['children']['linux'][details['distro']].keys() else Ydata['all']['children']['linux'][details['distro']]['port']
	msg = f"{host} services: {details['details']['services']}" if details is not None else f"Failed to obtain info for {host}"
	level = logging.INFO if details is not None else logging.WARNING
	FileTools.WriteLog(msg,log,level) if log is not None else print(msg)
	Security = AnsibleTools.AnsibleCredentials(Config['Args']['inventory'],Config['Args']['password'])
	return {'host':host,'key':key,'port':port,'admin':Security['sudo_user'],'password':Security['sudo_pass'],'log':log,'services':details['details']['services']}
# Function to get the Commandline arguments => Uses common initializer
def GetArguments():
	script_path = os.path.abspath(sys.argv[0])
	script_dir = os.path.dirname(script_path)
	script_name = os.path.basename(script_path)
	Config["Script"] = {"path":script_path,"folder":script_dir,"name":script_name}
	MyScript = f"{os.path.splitext(script_name)[0]}"
	parser = MyParser(
		prog=f"{MyScript}",
		description='Verifies that the specified services for the Specified Host has Started',
		epilog='Inventory Files: hosts.yml and vault.yml overrides all defaults\nexcept the ssh keys which can be locally defined',
		script=f"{MyScript}",
		script_version='1.0.0',
		include_python_version=True,
		ansible_host=True,
		usage = "%(prog)s [-h -v --dry-run --log-level -H -i -p --version -L -s] \
			\n(-h displays help, --version displays version)",
		formatter_class=argparse.RawTextHelpFormatter
	)
	parser.add_argument("-k", '--key', dest='key', type=str, help=f"locations of SSH Keys, defaults to vault specs")
	try:
		known,unknown = parser.parse_known_args()
		if '-h' in unknown or '--help' in unknown:
			help_text = parser.get_help_text()
			parser.print_help()
			sys.exit(2)
		args = vars(known)
		args['log']=Tools.String2Dictionary(args['log'],'=')
		args['inventory'] = os.path.join(Config['Script']['folder'],'Data') if args['inventory'] is None else args['inventory']
		args['password'] = os.path.expanduser(args['password']) if '~/' in args['password'] else args['password']
		args['password'] = AnsibleTools.read_vault_password(args['password'],args['log']['path']) if os.path.isfile(args['password']) else args['password']
		msg = None
		if args['key'] is None:
			tmp = AnsibleTools.AnsibleCredentials(args['inventory'],args['password'])
			args['key'] = tmp['key_path'] if tmp is not None and 'key_path' in tmp.keys() else None
		if msg is not None:
			FileTools.WriteLog(msg,args['log']['path']) if args['log']['path'] is not None else print(msg)
			raise argparse.ArgumentError(None,msg)
		return args
	except argparse.ArgumentError as e:
		parser.error(str(e))
# Function to extract suspicious addresses
def ExtractSuspectAddresses(log_path,patterns):
	rc = FileTools.CheckFileAccess(log_path,Config['Host']['log'])
	FileTools.WriteLog(f"Checking Access to {log_path} returned {rc}",Config['Host']['log'])
	if rc == 0:
		lines = FileTools.ReadTextFile(log_path)
	elif rc == 2:
		cmd = ["cat",log_path]
		rc = NetTools.Sudo_Run(cmd,Config['Host']['password'])
		lines = rc['out'].split() if 'out' in rc.keys() and rc['code'] == 0 else []
	ips = set()
	FileTools.WriteLog(f"Reading {log_path} returned {len(lines)} lines to be processed",Config['Host']['log'])
	for line in lines:
		for pattern in patterns:
			if re.search(pattern, line, re.IGNORECASE):
				print(f"{pattern}: {line}")
				match = re.search(r'client ([\d\.]+)', line)
				if match:
					ips.add(match.group(1))
	return ips

def extract_suspicious_ips(log_path, patterns):
    if not log_path.exists():
        raise FileNotFoundError(f"Log file not found: {log_path}")
    
    with log_path.open("r") as f:
        lines = f.readlines()
    
    ips = set()
    for line in lines:
        for pattern in patterns:
            if re.search(pattern, line, re.IGNORECASE):
                match = re.search(r'client ([\d\.]+)', line)
                if match:
                    ips.add(match.group(1))
    return ips
"""
def is_ip_blocked(ip, password):
    output = sudo_run(["firewall-cmd", "--list-rich-rules"], password)
    return ip in output

def block_ip(ip, password):
    if is_ip_blocked(ip, password):
        print(f"🔒 Already blocked: {ip}")
        return
    cmd = BLOCK_CMD_TEMPLATE.copy()
    cmd[3] = cmd[3].format(ip)
    sudo_run(cmd, password)
    print(f"🚫 Blocked IP: {ip}")

def main():
    password = get_sudo_password()
    bad_ips = extract_suspicious_ips(LOG_PATH, SUSPICIOUS_PATTERNS)
    for ip in bad_ips:
        block_ip(ip, password)
    if bad_ips:
        sudo_run(RELOAD_CMD, password)
"""
# Function to Find the local Common folder and append it to the library path (Common)
def InitializeModules():
	ScriptPath = pathlib.Path(__file__).parent.resolve()
	LibPath = f"{ScriptPath}/Lib"
	if not CheckSysPath(LibPath): sys.path.append(LibPath)
# Main Function
def main():
	Config['Args'] = GetArguments()
	FileTools.ArchiveLog(Config['Args']['log'])
	logging.basicConfig(filename=Config['Args']['log']['path'],filemode='a',level=logging.WARNING)
	Config['Host'] = ConfigureHosts(Config['Args'])
	print(Config)
	bad_ips = ExtractSuspectAddresses(LOG_PATH,SUSPICIOUS_PATTERNS)
	#args = {'log_path':LOG_PATH,'pattern':SUSPICIOUS_PATTERNS,'log':Config['Args']['log']['path'],'password':Config['Host']['password']}
	#bad_ips = NetTools.GetSuspicousAddresses(**args)
	print(bad_ips)
	sys.exit(0)
	for ip in bad_ips:
		block_ip(ip)
	if bad_ips:
		reload_firewalld()
# Preload with User Libraries -> Common to all scripts
InitializeModules()
import FileTools,NetTools,Tools,DbTools,AnsibleTools
from MyParser import MyParser
# Execute Script
if __name__ =="__main__":
    main()