check_isvpn

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

import os
import sys
import time
import subprocess
import socket
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)

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

original_getaddrinfo = socket.getaddrinfo

def forced_ipv6_gai_family(*args, **kwargs):
    global original_getaddrinfo
    responses = original_getaddrinfo(*args, **kwargs)
    return [response
            for response in responses
            if response[0] == socket.AF_INET6]

def forced_ipv4_gai_family(*args, **kwargs):
    global original_getaddrinfo
    responses = original_getaddrinfo(*args, **kwargs)
    return [response
            for response in responses
            if response[0] == socket.AF_INET]

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

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 doisvpncall(expected,severity,proxy):
    try:
        driver = generatedriver(proxy)

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

        infomsg("browser OK")
        url = "https://iphub.info/"
        driver.get(url)
        infomsg("asking for navigating "+url)

        time.sleep(5)
        wait_page_loaded(driver)

        wait.until(EC.presence_of_element_located((By.ID, "type")))

        element = driver.find_element(By.ID, "type")
        if element == None:
            exitnagios("CRITICAL","no result from page")

        value = element.text.strip()
        if value in ["It seems that this address is not a VPN", "Good IP (residential or business)"]:
            if expected:
                exitnagios(severity,"this is not a VPN, but it was expected to be")
            else:
                exitnagios("OK","this is not a VPN")
        elif value in ["It seems that this address is a VPN", "Hosting, proxy or bad IP"]:
            if expected:
                exitnagios("OK","this is a VPN as expected")
            else:
                exitnagios(severity,"this is unexpectedly a VPN")
        else:
            exitnagios("CRITICAL","cannot interpret output - "+value)
    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 - "+exceptionmsg)

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

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("-e", "--expected",                                dest="expected",         default="false",                              help="expected true/false")
    parser.add_argument("-s", "--severity",                                dest="severity",         default="CRITICAL",                           help="severity of failure")
    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
        socket.getaddrinfo = forced_ipv4_gai_family
    if args.ipv6 == True :
        protocol = "-6"
        exclist = exclist+1
        socket.getaddrinfo = forced_ipv6_gai_family
    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.expected.lower() in ["1","yes","true"]:
        expected = True
    else:
        expected = False
    proxy = ""

    if args.severity not in ["CRITICAL","WARNING"] :
        exitnagios("CRITICAL","the severity '"+args.severity+"' is not valid")

    doisvpncall(expected,args.severity,proxy)

if __name__ == "__main__":
    main()

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