check_dnsleak

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

import os
import sys
import time
import subprocess
import argparse

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.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 extractvalue(driver,entryname):
    element = driver.find_element(By.XPATH, "//td[text()=\""+entryname+"\"]/following-sibling::td")
    value = element.text
    infomsg(value)
    return value

def extracttable(driver):
    element = driver.find_element(By.ID, "dns-list")
    value = ""
    lines = element.find_elements(By.TAG_NAME,"tr")
    infomsg(element.text)
    for line in lines:
        if len(value)>0:
            value = value + " § "
        cells = line.find_elements(By.TAG_NAME,"td")
        entry = ""
        for cell in cells:
            if len(entry)>0:
                entry = entry + " ; "
            entry = entry + cell.text
        value = value + entry
    return value

def comparepart(value,expecteds):
    listexpected = expecteds.split("¶")
    result = False
    for alternative in listexpected:
        result = result or value==alternative.strip()
    return result

def comparetriplet(value,expected):
    if expected == "":
        return True
    else:
        listvalue = value.split(";")
        listexpected = expected.split(";")

        if (len(listvalue)==3) and (len(listexpected)==3):
            cleanvalue = []
            for value in listvalue:
                cleanvalue.append(value.strip())

            cleanexpected = []
            for expected in listexpected:
                cleanexpected.append(expected.strip())

            result = True
            for i in range(0,3):
                if cleanexpected[i]!="":
                    result = result and comparepart(cleanvalue[i],cleanexpected[i])
            return result
        else:
            return False

def parseip(inputvalue):
    result = []
    inputvalue = inputvalue.strip()
    if inputvalue!="":
        for item in inputvalue.split("§"):
            item = item.strip()
            if item not in result:
                result.append(item)
    return result

def parsedns(inputvalue):
    result = []
    inputvalue = inputvalue.strip()
    if inputvalue!="":
        for item in inputvalue.split("§"):
            item = item.strip()
            if item not in result:
                result.append(item)
    return result

def findtripletsmatching(inputtriplets,expectedtriplets):
    result = []
    for testinputtriplet in inputtriplets:
        for testexpectedtriplet in expectedtriplets:
            if comparetriplet(testinputtriplet.strip(),testexpectedtriplet.strip()):
                result.append(testinputtriplet)
    return result

def alltripletsmatching(inputtriplets,expectedtriplets,tolerance):
    successmatching = True
    failed = []
    tolerated = []
    for testinputtriplet in inputtriplets:
        partial = False
        for testexpectedtriplet in expectedtriplets:
            infomsg(testinputtriplet.strip())
            infomsg(testexpectedtriplet.strip())
            if comparetriplet(testinputtriplet.strip(),testexpectedtriplet.strip()):
                partial = True
                break
        infomsg(partial)
        if partial==False:
            failed.append(testinputtriplet.strip())
    if tolerance >= len(failed):
        tolerated = failed.copy()
        failed = []
    successmatching = (len(failed)==0)
    return successmatching,tolerated

def dodnsleakcall(proxy,problematicip,problematicdns,expectedip,expecteddns,tolerancedns):
    try:
        driver = generatedriver(proxy)

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

        infomsg("browser OK")
        url = "https://browserleaks.com/dns"
        driver.get(url)
        infomsg("asking for navigating "+url)

        time.sleep(5)
        wait_page_loaded(driver)
        infomsg("page loaded")

        counter=0
        retries=3
        while counter<retries:
            infomsg("retry "+str(counter))
            try:
                wait.until(EC.text_to_be_present_in_element((By.ID, "dns-test"),"Found"))
            except Exception as e:
                pass
            finally:
                infomsg("found element dns-test")
            try:
                resume = driver.find_element(By.ID, "dns-test")
            except Exception as e:
                exitnagios("WARNING","cannot find result element")
            if resume == None:
                counter=counter+1
                time.sleep(3)
            else:
                break

        if resume == None:
            exitnagios("WARNING","no result from page")
        infomsg("test with results")

        address = extractvalue(driver, "IP Address")
        isp = extractvalue(driver, "ISP")
        location = extractvalue(driver, "Location")
        resultedip = address.replace(";"," ")+" ; "+isp.replace(";"," ")+" ; "+location.replace(";"," ")
        infomsg(resultedip)

        resulteddns = ""
        try:
            wait.until(EC.presence_of_element_located((By.ID, "dns-list")))
            resulteddns = extracttable(driver).strip()
        except Exception as e:
            pass
        finally:
            infomsg("found element dns-list as '"+resulteddns+"'")

        successmsg = ""
        warningmsg = ""
        errormsg = ""

        if resultedip == "":
            warningmsg += "The IP response is empty, and was expected '"+" § ".join(expectedip)+"'. "
        else:
            # if something problematic is defined, find it and report it,
            # ifnot,... all the resulted dns should be in the expected dns
            resultedipparts = parseip(resultedip)
            infomsg(resultedipparts)
            if len(problematicip)>0:
                failureip = findtripletsmatching(resultedipparts,problematicip)
                if len(failureip)>0:
                    errormsg += "The IP response '"+" § ".join(resultedipparts)+"' is problematic. "
                else:
                    successmsg += "The IP response '"+" § ".join(resultedipparts)+"' is not problematic. "
            else:
                successip,toleratedip = alltripletsmatching(resultedipparts,expectedip,0)
                if successip or len(expectedip)==0:
                    successmsg += "The IP response '"+" § ".join(resultedipparts)+"' is as expected. "
                else:
                    errormsg += "The IP response '"+" § ".join(resultedipparts)+"' is not as expected '"+" § ".join(expectedip)+"'. "

        if resulteddns == "":
            warningmsg += "The DNS response is empty, and was expected '"+" § ".join(expecteddns)+"'. "
        else:
            # if something problematic is defined, find it and report it,
            # ifnot,... all the resulted dns should be in the expected dns
            resulteddnsparts = parsedns(resulteddns)
            infomsg(resulteddnsparts)
            if len(problematicdns)>0:
                failuredns = findtripletsmatching(resulteddnsparts,problematicdns)
                if len(failuredns)>tolerancedns:
                    errormsg += "The DNS response '"+" § ".join(resulteddnsparts)+"' is problematic. "
                elif len(failuredns)>0:
                    warningmsg += "The DNS response '"+" § ".join(resulteddnsparts)+"' is tolerated. "
                else:
                    successmsg += "The DNS response '"+" § ".join(resulteddnsparts)+"' is not problematic. "
            else:
                successdns,tolerateddns = alltripletsmatching(resulteddnsparts,expecteddns,tolerancedns)
                if successdns or len(expecteddns)==0:
                    if len(tolerateddns)==0:
                        successmsg += "The DNS response '"+" § ".join(resulteddnsparts)+"' is as expected. "
                    else:
                        warningmsg += "The DNS response '"+" § ".join(resulteddnsparts)+"' is tolerated. "
                else:
                    errormsg += "The DNS response '"+" § ".join(resulteddnsparts)+"' is not as expected '"+" § ".join(expecteddns)+"'. "

        if errormsg != "":
            exitnagios("CRITICAL",errormsg)
        elif warningmsg != "":
            exitnagios("WARNING",warningmsg)
        else:
            exitnagios("OK",successmsg)
    except Exception as e:
        exceptionmsg = str(e)
        if (exceptionmsg.find("net::ERR_CONNECTION_REFUSED") > -1) or (exceptionmsg.find("net::ERR_SOCKS_CONNECTION_FAILED") > -1) or (exceptionmsg.find("net::ERR_ADDRESS_UNREACHABLE") > -1):
            exitnagios("WARNING","no connection")
        else:
            exitnagios("CRITICAL","unexpected error during the check - "+str(e))

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

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--proxy",                                   dest="proxy",            default="",                                   help="use a proxy")
    parser.add_argument("-x", "--problematicip",                           dest="problematicip",    default="",                                   help="raise critical if this IP is found")
    parser.add_argument("-r", "--problematicdns",                          dest="problematicdns",   default="",                                   help="raise critical if this DNS is found")
    parser.add_argument("-i", "--expectedip",                              dest="expectedip",       default="",                                   help="expected response for the IP")
    parser.add_argument("-d", "--expecteddns",                             dest="expecteddns",      default="",                                   help="expected response for the DNS")
    parser.add_argument("-t", "--tolerancedns",                            dest="tolerancedns",     default="0",                                  help="number of problematic DNS entries to tolerate")
    parser.add_argument("-®", "--debug",             action="store_true",  dest="debug",            default=False,                                help="be more verbose")
    args = parser.parse_args()

    setcandebug(args.debug)

    problematicip = parseip(args.problematicip)
    infomsg(problematicip)
    problematicdns = parsedns(args.problematicdns)
    infomsg(problematicdns)
    expectedip = parseip(args.expectedip)
    infomsg(expectedip)
    expecteddns = parsedns(args.expecteddns)
    infomsg(expecteddns)
    tolerancedns = int(args.tolerancedns)
    infomsg(tolerancedns)

    if ((len(problematicip)!=0) and (len(expectedip)!=0)) or ((len(problematicdns)!=0) and (len(expecteddns)!=0)):
        exitnagios("CRITICAL","problematic and expected flags are mutually exclusive")

    if (problematicip=="") and (problematicdns=="") and (problematicip=="") and (problematicdns==""):
        exitnagios("CRITICAL","a problematic or expected flat must be specified")

    dodnsleakcall(args.proxy,problematicip,problematicdns,expectedip,expecteddns,tolerancedns)

if __name__ == "__main__":
    main()

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