check_emailblacklists

#!/usr/bin/env python3
#-------------------------------------------------------------------------------

import os
import sys
import time
import subprocess
import argparse
import datetime
import dns.resolver

from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.options import Options
from selenium.webdriver import ActionChains

from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

#-------------------------------------------------------------------------------

_candebug = False
def candebug():
    global _candebug
    return _candebug
def setcandebug(value):
    global _candebug
    _candebug = value

def infomsg(msg):
    if candebug() == True:
        print(msg, flush=True)

def exitnagios(status,message):
    if status=="OK":
        exitcode = 0
    elif status=="WARNING":
        exitcode = 1
    elif status=="CRITICAL":
        exitcode = 2
    elif status=="UNKNOWN":
        exitcode = 3
    else:
        exitcode = 4
    print(status+": "+message, flush=True)
    sys.exit(exitcode)

#-------------------------------------------------------------------------------

def generatedriver(proxy):
    chrome_options = Options()
    chrome_options.add_argument("--incognito")
    chrome_options.add_argument("--disk-cache-size=0")
    chrome_options.add_argument("--media-cache-size=0")
    chrome_options.add_argument("--enable-features=EphemeralGuestProfilesOnStartup")
    chrome_options.add_argument("--headless=new")
    chrome_options.add_argument("--disable-gpu")
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-dev-shm-usage")
    chrome_options.add_argument("--in-process-gpu")
    chrome_options.add_argument("--use-gl=angle")
    chrome_options.add_argument("--use-angle=swiftshader-webgl")
    chrome_options.add_argument("--proxy-server="+proxy)

    chrome_service = webdriver.ChromeService(executable_path="/usr/bin/chromedriver")

    driver = webdriver.Chrome(options=chrome_options,service=chrome_service)
    #driver.implicitly_Wait(10, TimeUnit.SECONDS)

    return driver

def check_page_loaded(driver):
    #self.log.info("Checking if {} page is loaded.".format(self.driver.current_url))
    page_state = driver.execute_script("return document.readyState;")
    return page_state == "complete"

def wait_page_loaded(driver):
    timestamp = time.time()
    while True:
        if check_page_loaded(driver) == True:
            break
        if time.time() - timestamp > 20:
            break

#-------------------------------------------------------------------------------

def tryfindbyxpath(driver,xpath):
    result = []
    try:
        result = driver.find_elements(By.XPATH, xpath)
    except:
        pass
    return result

def tryfindbyid(driver,idname):
    result = []
    try:
        result = driver.find_elements(By.ID, idname)
    except:
        pass
    return result

def tryfindbyclassname(driver,classname):
    result = []
    try:
        result = driver.find_elements(By.CLASS_NAME, classname)
    except:
        pass
    return result

def tryfindbytagname(driver,tagname):
    result = []
    try:
        result = driver.find_elements(By.TAG_NAME, tagname)
    except:
        pass
    return result

#-------------------------------------------------------------------------------

def resolverecord(protocol,hostname):
    result = []
    try:
        record = None
        if protocol in ["-4","4"]:
            record = "A"
        if protocol in ["-6","6"]:
            record = "AAAA"
        if record!=None:
            answers = dns.resolver.resolve(hostname, record)
            for rdata in answers:
                result.append(rdata.to_text())
    except Exception as e:
        infomsg("exception "+str(e))
    return result

def dodnssecverisigncall(protocol,hostname,ignorelist,tolerance,proxy):
    try:
        ips = resolverecord(protocol,hostname)
        if len(ips)==0:
            exitnagios("CRITICAL","cannot resolve "+hostname+" with protocol "+protocol)
        if len(ips)>1:
            exitnagios("CRITICAL","too many ips resolved for "+hostname+" - "+str(ips))
        targetip = ips[0]
        infomsg("The IP to test is "+targetip)

        driver = generatedriver(proxy)

        wait = WebDriverWait(driver,20)
        action=ActionChains(driver)

        # OTHER ALTERNATIVES
        #https://mxtoolbox.com/SuperTool.aspx?action=blacklist
        #https://www.dnsbl.info/dnsbl-database-check.php
        #https://blacklistchecker.com/check
        #https://www.uceprotect.net/en/rblcheck.php
        
        # OTHER OPTION
        # https://manpages.debian.org/bookworm/amispammer/amispammer.1.en.html -- perl code gives warnings, and fails with IPv6

        infomsg("browser OK")
        url = "https://dnschecker.org/ip-blacklist-checker.php"
        driver.get(url)
        infomsg("asking for navigating "+url)

        infomsg("manual delay")
        time.sleep(2)

        wait_page_loaded(driver)
        infomsg("page loaded")

        infomsg("waiting for submit")
        wait.until(EC.presence_of_element_located((By.ID, "ipbc_submit")))

        infomsg("finding textbox element")
        elements = tryfindbyid(driver,"ipbc_host")
        if len(elements)==0:
            exitnagios("CRITICAL","cannot find textbox")
        else:
            infomsg("sending ip")
            elements[0].send_keys(targetip)
            elements[0].send_keys(Keys.RETURN)

        infomsg("manual delay")
        time.sleep(10)

        infomsg("waiting for result")
        wait.until(EC.presence_of_element_located((By.ID, "blacklist_count")))

        infomsg("extracting listed")
        blocks = tryfindbyid(driver,"ipbc_results_dnsbl")
        lists = tryfindbytagname(blocks[0],"tr")
        infomsg("row canidates: "+str(len(lists)))
        issues=[]
        for item in lists:
            providers = tryfindbytagname(item,"th")
            if len(providers)==1:
                provider = providers[0].text
            else:
                provider = ""
            statuses = tryfindbytagname(item,"td")
            if len(statuses)==2:
                listed = statuses[0].text.split()[0]
            else:
                listed = ""
            if provider!="" and listed!="":
                infomsg("status "+provider+" "+listed)
                if listed.lower()=="yes":
                    if provider in ignorelist:
                        infomsg("ignoring "+provider)
                    else:
                        infomsg("adding "+provider)
                        issues.append(provider)

        if len(issues)>tolerance:
            exitnagios("CRITICAL","The IP "+targetip+" is listed in spam databases "+str(issues)+" | count="+str(len(issues)))
        elif len(issues)>0:
            exitnagios("WARNING","The IP "+targetip+" is listed in spam databases "+str(issues)+" | count="+str(len(issues)))
        else:
            exitnagios("OK","No issues found for the IP "+targetip+" | count="+str(len(issues)))

    except Exception as e:
        exitnagios("CRITICAL","unexpected error during the check "+str(e))

#-------------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-4", "--ipv4",              action="store_true",  dest="ipv4",             default=False,                                help="use ipv4 (exclusive with ipv6)")
    parser.add_argument("-6", "--ipv6",              action="store_true",  dest="ipv6",             default=False,                                help="use ipv6 (exclusive with ipv4)")
    parser.add_argument("-0", "--ipv0",              action="store_true",  dest="ipv0",             default=False,                                help="do not specify ip protocol version")
    parser.add_argument("-H", "--hostname",                                dest="hostname",                                                       help="server to check")
    parser.add_argument("-i", "--ignore",                                  dest="ignore",           default="",                                   help="lists to ignore")
    parser.add_argument("-l", "--tolerance",                               dest="tolerance",        default="0",                                  help="number of listed databases to tolerate")
    parser.add_argument("-®", "--debug",             action="store_true",  dest="debug",            default=False,                                help="be more verbose")
    args = parser.parse_args()

    setcandebug(args.debug)

    protocol = None
    exclist = 0
    if args.ipv0 == True :
        protocol = ""
        exclist = exclist+1
    if args.ipv4 == True :
        protocol = "-4"
        exclist = exclist+1
    if args.ipv6 == True :
        protocol = "-6"
        exclist = exclist+1
    if (exclist>1) :
        exitnagios("CRITICAL","select ip protocol version 4 or 6, or 0 to relay on OS handling")
    elif (exclist<1) :
        exitnagios("CRITICAL","no protocol selected")

    if (args.hostname == ""):
        exitnagios("CRITICAL","hostname not defined")

    proxy = ""
    ignorelist = args.ignore.split(",")
    tolerance = int(args.tolerance)
    dodnssecverisigncall(protocol,args.hostname,ignorelist,tolerance,proxy)

if __name__ == "__main__":
    main()

#-------------------------------------------------------------------------------