diff --git a/gp-okta.conf b/gp-okta.conf index d14ad64..73af663 100644 --- a/gp-okta.conf +++ b/gp-okta.conf @@ -1,15 +1,27 @@ debug = 0 + vpn_url = https://vpn.example.com +#vpn_url_cert = vpn_url.cert + okta_url = https://example.okta.com +#okta_url_cert = okta_url.cert + username = myuser password = mypass +#client_cert = path-to-myusers-client-cert-as-unencrypted-pem-file.pem + # mfa_order = totp sms sms.okta = 0 -totp.okta = ABCDEFGHIJKLMNOP -totp.google = ABCDEFGHIJKLMNOP -gateway = Manual ny1-gw.example.com +#totp.okta = ABCDEFGHIJKLMNOP +#totp.google = ABCDEFGHIJKLMNOP +#totp.symantec = ABCDEFGHIJKLMNOP + +gateway = Manual ny1-gw.example.com # optional hardcoded gateway + #openconnect_cmd = sudo openconnect +#openconnect_certs = path-to-a-writeable-filename-in-which-the-script-will-collect-all-involved-server-certs-if-not-set-a-temp-file-is-used-instead openconnect_args = # optional arguments to openconnect execute = 0 # execute openconnect command +another_dance = 0 # second round of authentication required bug.nl = 0 # newline work-around for openconnect bug.username = 0 # username work-around for openconnect diff --git a/gp-okta.py b/gp-okta.py index 2ccba41..b83cb69 100755 --- a/gp-okta.py +++ b/gp-okta.py @@ -2,19 +2,24 @@ # -*- coding: utf-8 -*- """ The MIT License (MIT) - + Copyright (C) 2018 Andris Raugulis (moo@arthepsy.eu) - + Copyright (C) 2018 Nick Lanham (nick@afternight.org) + Copyright (C) 2019 Aaron Lindsay (aclindsa@gmail.com) + Copyright (C) 2019 Taylor Dean (taylor@makeshift.dev) + Copyright (C) 2020 Max Lanin (mlanin@evolutiongaming.com) + Copyright (C) 2019-2020 Tino Lange (coldcoff@yahoo.com) + Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: - + The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. - + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE @@ -23,38 +28,70 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ -from __future__ import print_function -import io, os, sys, re, json, base64, getpass, subprocess, shlex, signal +from __future__ import print_function, unicode_literals +import io, os, sys, time, re, json, base64, getpass, subprocess, shlex, signal, tempfile, traceback + from lxml import etree import requests +import argparse if sys.version_info >= (3,): - from urllib.parse import urljoin + from urllib.parse import urlparse, urljoin text_type = str binary_type = bytes else: - from urlparse import urljoin + from urlparse import urlparse, urljoin text_type = unicode binary_type = str input = raw_input +# Optional: fido2 support (webauthn via Yubikey) +have_fido = False +try: + from fido2.utils import websafe_decode, websafe_encode + from fido2.hid import CtapHidDevice + from fido2.client import Fido2Client + have_fido = True +except ImportError: + pass + +# Optional: pyotp support +have_pyotp = False +try: + import pyotp + have_pyotp = True +except ImportError: + pass + +# Optional: gnupg support +have_gnupg = False +try: + import gnupg + have_gnupg = True +except ImportError: + pass + to_b = lambda v: v if isinstance(v, binary_type) else v.encode('utf-8') to_u = lambda v: v if isinstance(v, text_type) else v.decode('utf-8') +quiet = False def log(s): - print('[INFO] {0}'.format(s)) + if not quiet: + print('[INFO] {0}'.format(s)) def dbg(d, h, *xs): + if quiet: + return if not d: return - print('# {0}:'.format(h)) + print('[DEBUG] {0}:'.format(h)) for x in xs: - print(x) - print('---') + print('[DEBUG] {0}'.format(x)) + print('[DEBUG] ---') def err(s): - print('err: {0}'.format(s)) + print('[ERROR] {0}'.format(s), file=sys.stderr) sys.exit(1) def parse_xml(xml): @@ -62,21 +99,21 @@ def parse_xml(xml): xml = bytes(bytearray(xml, encoding='utf-8')) parser = etree.XMLParser(ns_clean=True, recover=True) return etree.fromstring(xml, parser) - except: - err('failed to parse xml') + except Exception as e: + err('failed to parse xml: ' + e) def parse_html(html): try: parser = etree.HTMLParser() return etree.fromstring(html, parser) - except: - err('failed to parse html') + except Exception as e: + err('failed to parse html: ' + e) def parse_rjson(r): try: return r.json() - except: - err('failed to parse json') + except Exception as e: + err('failed to parse json: ' + e) def parse_form(html, current_url = None): xform = html.find('.//form') @@ -91,25 +128,26 @@ def parse_form(html, current_url = None): data[k] = v return url, data - def load_conf(cf): + log('load conf') conf = {} keys = ['vpn_url', 'username', 'password', 'okta_url'] + if isinstance(cf, binary_type): + cf = cf.decode('utf-8') line_nr = 0 - with io.open(cf, 'r', encoding='utf-8') as fp: - for rline in fp: - line_nr += 1 - line = rline.strip() - mx = re.match(r'^\s*([^=\s]+)\s*=\s*(.*?)\s*(?:#\s+.*)?\s*$', line) - if mx: - k, v = mx.group(1).lower(), mx.group(2) - if k.startswith('#'): - continue - for q in '"\'': - if re.match(r'^{0}.*{0}$'.format(q), v): - v = v[1:-1] - conf[k] = v - conf['{0}.line'.format(k)] = line_nr + for rline in cf.split('\n'): + line_nr += 1 + line = rline.strip() + mx = re.match(r'^\s*([^=\s]+)\s*=\s*(.*?)\s*(?:#\s+.*)?\s*$', line) + if mx: + k, v = mx.group(1).lower(), mx.group(2) + if k.startswith('#'): + continue + for q in '"\'': + if re.match(r'^{0}.*{0}$'.format(q), v): + v = v[1:-1] + conf[k] = v + conf['{0}.line'.format(k)] = line_nr for k, v in os.environ.items(): k = k.lower() if k.startswith('gp_'): @@ -121,6 +159,22 @@ def load_conf(cf): conf['username'] = input('username: ').strip() if len(conf.get('password', '').strip()) == 0: conf['password'] = getpass.getpass('password: ').strip() + if len(conf.get('openconnect_certs', '').strip()) == 0: + conf['openconnect_certs'] = tempfile.NamedTemporaryFile() + log('will collect openconnect_certs in temporary file: {0} and verify against them'.format(conf['openconnect_certs'].name)) + else: + conf['openconnect_certs'] = open(conf['openconnect_certs'], 'wb') + if len(conf.get('vpn_url_cert', '').strip()) != 0: + if not os.path.exists(conf.get('vpn_url_cert')): + err('configured vpn_url_cert file does not exist') + conf['openconnect_certs'].write(open(conf.get('vpn_url_cert'), 'rb').read()) # copy it + if len(conf.get('okta_url_cert', '').strip()) != 0: + if not os.path.exists(conf.get('okta_url_cert')): + err('configured okta_url_cert file does not exist') + conf['openconnect_certs'].write(open(conf.get('okta_url_cert'), 'rb').read()) # copy it + if len(conf.get('client_cert', '').strip()) != 0: + if not os.path.exists(conf.get('client_cert')): + err('configured client_cert file does not exist') for k in keys: if k not in conf: err('missing configuration key: {0}'.format(k)) @@ -128,12 +182,13 @@ def load_conf(cf): if len(conf[k].strip()) == 0: err('empty configuration key: {0}'.format(k)) conf['debug'] = conf.get('debug', '').lower() in ['1', 'true'] + conf['openconnect_certs'].flush() return conf def mfa_priority(conf, ftype, fprovider): - if ftype == 'token:software:totp': + if ftype == 'token:software:totp' or (ftype, fprovider) == ('token', 'symantec'): ftype = 'totp' - if ftype not in ['totp', 'sms']: + if ftype not in ['totp', 'sms', 'push', 'webauthn']: return 0 mfa_order = conf.get('mfa_order', '') if ftype in mfa_order: @@ -141,7 +196,7 @@ def mfa_priority(conf, ftype, fprovider): else: priority = 0 value = conf.get('{0}.{1}'.format(ftype, fprovider)) - if ftype == 'sms': + if ftype in ('sms', 'webauthn'): if not (value or '').lower() in ['1', 'true']: value = None line_nr = conf.get('{0}.{1}.line'.format(ftype, fprovider), 0) @@ -153,6 +208,13 @@ def mfa_priority(conf, ftype, fprovider): priority += (512 - line_nr) return priority +def get_state_token(conf, c, current_url = None): + rx_state_token = re.search(r'var\s*stateToken\s*=\s*\'([^\']+)\'', c) + if not rx_state_token: + dbg(conf.get('debug'), 'not found', 'stateToken') + return None + state_token = to_b(rx_state_token.group(1)).decode('unicode_escape').strip() + return state_token def get_redirect_url(conf, c, current_url = None): rx_base_url = re.search(r'var\s*baseUrl\s*=\s*\'([^\']+)\'', c) @@ -173,6 +235,13 @@ def get_redirect_url(conf, c, current_url = None): def send_req(conf, s, name, url, data, **kwargs): dbg(conf.get('debug'), '{0}.request'.format(name), url) + if kwargs.get('expected_url'): + purl = list(urlparse(url)) + purl = (purl[0], purl[1].split(':')[0]) + pexp = list(urlparse(kwargs.get('expected_url'))) + pexp = (pexp[0], pexp[1].split(':')[0]) + if purl != pexp: + err('{0}: unexpected url found {1} != {2}'.format(name, purl, pexp)) do_json = True if kwargs.get('json') else False headers = {} if do_json: @@ -180,48 +249,67 @@ def send_req(conf, s, name, url, data, **kwargs): headers['Accept'] = 'application/json' headers['Content-Type'] = 'application/json' if kwargs.get('get'): - r = s.get(url, headers=headers) + r = s.get(url, headers=headers, + verify=kwargs.get('verify', True)) else: - r = s.post(url, data=data, headers=headers) + r = s.post(url, data=data, headers=headers, + verify=kwargs.get('verify', True)) hdump = '\n'.join([k + ': ' + v for k, v in sorted(r.headers.items())]) rr = 'status: {0}\n\n{1}\n\n{2}'.format(r.status_code, hdump, r.text) if r.status_code != 200: - err('okta {0} request failed. {0}'.format(rr)) + err('{0}.request failed.\n{1}'.format(name, rr)) dbg(conf.get('debug'), '{0}.response'.format(name), rr) if do_json: return r.headers, parse_rjson(r) return r.headers, r.text - -def paloalto_prelogin(conf, s): - log('prelogin request') - url = '{0}/global-protect/prelogin.esp'.format(conf.get('vpn_url')) - h, c = send_req(conf, s, 'prelogin', url, {}, get=True) +def paloalto_prelogin(conf, s, gateway=None): + verify = None + if gateway: + # 2nd round: use gateway + log('prelogin request [gateway]') + url = 'https://{0}/ssl-vpn/prelogin.esp'.format(gateway) + verify = conf.get('vpn_url_cert') + else: + # 1st round: use portal + log('prelogin request [vpn_url]') + url = '{0}/global-protect/prelogin.esp'.format(conf.get('vpn_url')) + if conf.get('openconnect_certs') and os.path.getsize(conf.get('openconnect_certs').name) > 0: + verify = conf.get('openconnect_certs').name + _h, c = send_req(conf, s, 'prelogin', url, {}, get=True, verify=verify) x = parse_xml(c) saml_req = x.find('.//saml-request') if saml_req is None: - err('did not find saml request') + msg = x.find('.//msg') + if msg is not None: + msg = msg.text + if msg is not None: + msg = msg.strip() + else: + msg = 'Probably SAML is disabled at the portal? Or you need a certificate? Try another_dance=0 with some concrete gateway instead.' + err('did not find saml request. {0}'.format(msg)) if len(saml_req.text.strip()) == 0: err('empty saml request') try: saml_raw = base64.b64decode(saml_req.text) - except: - err('failed to decode saml request') + except Exception as e: + err('failed to decode saml request: ' + e) dbg(conf.get('debug'), 'prelogin.decoded', saml_raw) saml_xml = parse_html(saml_raw) return saml_xml def okta_saml(conf, s, saml_xml): - log('okta saml request') + log('okta saml request [okta_url]') url, data = parse_form(saml_xml) - h, c = send_req(conf, s, 'saml', url, data) + _h, c = send_req(conf, s, 'saml', url, data, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) redirect_url = get_redirect_url(conf, c, url) if redirect_url is None: err('did not find redirect url') return redirect_url -def okta_auth(conf, s): - log('okta auth request') +def okta_auth(conf, s, stateToken = None): + log('okta auth request [okta_url]') url = '{0}/api/v1/authn'.format(conf.get('okta_url')) data = { 'username': conf.get('username'), @@ -230,8 +318,11 @@ def okta_auth(conf, s): 'warnBeforePasswordExpired':True, 'multiOptionalFactorEnroll':True } + } if stateToken is None else { + 'stateToken': stateToken } - h, j = send_req(conf, s, 'auth', url, data, json=True) + _h, j = send_req(conf, s, 'auth', url, data, json=True, + verify=conf.get('okta_url_cert')) while True: ok, r = okta_transaction_state(conf, s, j) @@ -254,7 +345,8 @@ def okta_transaction_state(conf, s, j): if len(state_token) == 0: err('empty state token') data = {'stateToken': state_token} - h, j = send_req(conf, s, 'skip', url, data, json=True) + _h, j = send_req(conf, s, 'skip', url, data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) return False, j # status: password_expired # status: recovery @@ -302,12 +394,17 @@ def okta_mfa(conf, s, j): if len(factors) == 0: err('no factors found') for f in sorted(factors, key=lambda x: x.get('priority', 0), reverse=True): - print(f) + #print(f) ftype = f.get('type') - if ftype == 'token:software:totp': + fprovider = f.get('provider') + if ftype == 'token:software:totp' or (ftype, fprovider) == ('token', 'symantec'): r = okta_mfa_totp(conf, s, f, state_token) elif ftype == 'sms': r = okta_mfa_sms(conf, s, f, state_token) + elif ftype == 'push': + r = okta_mfa_push(conf, s, f, state_token) + elif ftype == 'webauthn': + r = okta_mfa_webauthn(conf, s, f, state_token) else: r = None if r is not None: @@ -321,7 +418,8 @@ def okta_mfa_totp(conf, s, factor, state_token): if len(secret) == 0: code = input('{0} TOTP: '.format(provider)).strip() else: - import pyotp + if not have_pyotp: + err('Need pyotp package, consider doing \'pip install pyotp\' (or similar)') totp = pyotp.TOTP(secret) code = totp.now() code = code or '' @@ -332,23 +430,93 @@ def okta_mfa_totp(conf, s, factor, state_token): 'stateToken': state_token, 'passCode': code } - log('mfa {0} totp request'.format(provider)) - h, j = send_req(conf, s, 'totp mfa', factor.get('url'), data, json=True) + log('mfa {0} totp request: {1} [okta_url]'.format(provider, code)) + _h, j = send_req(conf, s, 'totp mfa', factor.get('url'), data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) return j def okta_mfa_sms(conf, s, factor, state_token): provider = factor.get('provider', '') data = { 'factorId': factor.get('id'), - 'stateToken': state_token, + 'stateToken': state_token } - log('mfa {0} sms request'.format(provider)) - h, j = send_req(conf, s, 'sms mfa', factor.get('url'), data, json=True) + log('mfa {0} sms request [okta_url]'.format(provider)) + _h, j = send_req(conf, s, 'sms mfa (1)', factor.get('url'), data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) code = input('{0} SMS verification code: '.format(provider)).strip() if len(code) == 0: return None data['passCode'] = code - h, j = send_req(conf, s, 'sms mfa', factor.get('url'), data, json=True) + log('mfa {0} sms request [okta_url]'.format(provider)) + _h, j = send_req(conf, s, 'sms mfa (2)', factor.get('url'), data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) + return j + +def okta_mfa_push(conf, s, factor, state_token): + provider = factor.get('provider', '') + data = { + 'factorId': factor.get('id'), + 'stateToken': state_token, + } + log('mfa {0} push request [okta_url]'.format(provider)) + status = 'MFA_CHALLENGE' + counter = 0 + while status == 'MFA_CHALLENGE': + _h, j = send_req(conf, s, 'push mfa ({0})'.format(counter), + factor.get('url'), data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) + status = j.get('status', '').strip() + dbg(conf.get('debug'), 'status', status) + if status == 'MFA_CHALLENGE': + time.sleep(3.33) + counter += 1 + return j + +def okta_mfa_webauthn(conf, s, factor, state_token): + if not have_fido: + err('Need fido2 package(s) for webauthn. Consider doing `pip install fido2` (or similar)') + devices = list(CtapHidDevice.list_devices()) + if not devices: + err('webauthn configured, but no U2F devices found') + return None + provider = factor.get('provider', '') + log('mfa {0} challenge request [okta_url]'.format(provider)) + data = { + 'stateToken': state_token + } + _h, j = send_req(conf, s, 'webauthn mfa challenge', factor.get('url'), data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) + factor = j['_embedded']['factor'] + profile = factor['profile'] + purl = list(urlparse(conf.get('okta_url'))) + rpid = purl[1].split(':')[0] + origin = '{0}://{1}'.format(purl[0], rpid) + challenge = factor['_embedded']['challenge']['challenge'] + credentialId = websafe_decode(profile['credentialId']) + allow_list = [{'type': 'public-key', 'id': credentialId}] + for dev in devices: + client = Fido2Client(dev, origin) + print('!!! Touch the flashing U2F device to authenticate... !!!') + try: + result = client.get_assertion(rpid, challenge, allow_list) + dbg(conf.get('debug'), 'assertion.result', result) + break + except: + traceback.print_exc(file=sys.stderr) + result = None + if not result: + return None + assertion, client_data = result[0][0], result[1] # only one cred in allowList, so only one response. + data = { + 'stateToken': state_token, + 'clientData': to_u(base64.b64encode(client_data)), + 'signatureData': to_u(base64.b64encode(assertion.signature)), + 'authenticatorData': to_u(base64.b64encode(assertion.auth_data)) + } + log('mfa {0} signature request [okta_url]'.format(provider)) + _h, j = send_req(conf, s, 'uf2 mfa signature', j['_links']['next']['href'], data, json=True, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) return j def okta_redirect(conf, s, session_token, redirect_url): @@ -366,17 +534,23 @@ def okta_redirect(conf, s, session_token, redirect_url): 'redirectUrl': redirect_url } url = '{0}/login/sessionCookieRedirect'.format(conf.get('okta_url')) - log('okta redirect request') - h, c = send_req(conf, s, 'redirect', url, data) + log('okta redirect request {0} [okta_url]'.format(rc)) + h, c = send_req(conf, s, 'redirect', url, data, + verify=conf.get('okta_url_cert')) + state_token = get_state_token(conf, c, url) redirect_url = get_redirect_url(conf, c, url) if redirect_url: form_url, form_data = None, {} else: xhtml = parse_html(c) form_url, form_data = parse_form(xhtml, url) + if state_token is not None: + log('stateToken: {0}'.format(state_token)) + okta_auth(conf, s, state_token) elif form_url: - log('okta redirect form request') - h, c = send_req(conf, s, 'redirect form', form_url, form_data) + log('okta redirect form request [vpn_url]') + h, c = send_req(conf, s, 'redirect form', form_url, form_data, + expected_url=conf.get('vpn_url'), verify=conf.get('vpn_url_cert')) saml_username = h.get('saml-username', '').strip() prelogin_cookie = h.get('prelogin-cookie', '').strip() if saml_username and prelogin_cookie: @@ -385,23 +559,30 @@ def okta_redirect(conf, s, session_token, redirect_url): dbg(conf.get('debug'), 'saml prop', [saml_auth_status, saml_slo]) return saml_username, prelogin_cookie -def paloalto_getconfig(conf, s, saml_username, prelogin_cookie): - log('getconfig request') +def paloalto_getconfig(conf, s, username = None, prelogin_cookie = None): + log('getconfig request [vpn_url]') url = '{0}/global-protect/getconfig.esp'.format(conf.get('vpn_url')) data = { - 'user': saml_username, - 'passwd': '', - 'inputStr': '', + #'jnlpReady': 'jnlpReady', + #'ok': 'Login', + #'direct': 'yes', 'clientVer': '4100', + #'prot': 'https:', 'clientos': 'Windows', - 'clientgpversion': '4.1.0.98', - 'computer': 'DESKTOP', 'os-version': 'Microsoft Windows 10 Pro, 64-bit', + #'server': '', + 'computer': 'DESKTOP', + #'preferred-ip': '', + 'inputStr': '', + 'user': username or conf['username'], + 'passwd': '' if prelogin_cookie else conf['password'], + 'clientgpversion': '4.1.0.98', # 'host-id': '00:11:22:33:44:55' - 'prelogin-cookie': prelogin_cookie, + 'prelogin-cookie': prelogin_cookie or '', 'ipv6-support': 'yes' } - h, c = send_req(conf, s, 'getconfig', url, data) + _h, c = send_req(conf, s, 'getconfig', url, data, + verify=conf.get('vpn_url_cert')) x = parse_xml(c) xtmp = x.find('.//portal-userauthcookie') if xtmp is None: @@ -409,43 +590,171 @@ def paloalto_getconfig(conf, s, saml_username, prelogin_cookie): portal_userauthcookie = xtmp.text if len(portal_userauthcookie) == 0: err('empty portal_userauthcookie') - return portal_userauthcookie + gateways = set() + xtmp = x.find('.//gateways//external//list') + if xtmp is not None: + for entry in xtmp: + gateways.add(entry.get('name')) + xtmp = x.find('.//root-ca') + if xtmp is not None: + for entry in xtmp: + cert = entry.find('.//cert').text + conf['openconnect_certs'].write(to_b(cert)) + conf['openconnect_certs'].flush() + return portal_userauthcookie, gateways +# Combined first half of okta_saml with second half of okta_redirect +def okta_saml_2(conf, s, gateway, saml_xml): + log('okta saml request (2) [okta_url]') + url, data = parse_form(saml_xml) + h, c = send_req(conf, s, 'okta saml request (2)', url, data, + expected_url=conf.get('okta_url'), verify=conf.get('okta_url_cert')) + xhtml = parse_html(c) + url, data = parse_form(xhtml) + + log('okta redirect form request (2) [gateway]') + verify = None + if conf.get('openconnect_certs') and os.path.getsize(conf.get('openconnect_certs').name) > 0: + verify = conf.get('openconnect_certs').name + h, c = send_req(conf, s, 'okta redirect form (2)', url, data, + expected_url='https://{0}'.format(gateway), verify=verify) + saml_username = h.get('saml-username', '').strip() + if len(saml_username) == 0: + err('saml-username empty') + prelogin_cookie = h.get('prelogin-cookie', '').strip() + if len(prelogin_cookie) == 0: + err('prelogin-cookie empty') + + return saml_username, prelogin_cookie def main(): - if len(sys.argv) < 2: - print('usage: {0} '.format(sys.argv[0])) - sys.exit(1) - conf = load_conf(sys.argv[1]) - + + parser = argparse.ArgumentParser(description=""" + This is an OpenConnect wrapper script that automates connecting to a + PaloAlto Networks GlobalProtect VPN using Okta 2FA.""") + + parser.add_argument('conf_file', + help='e.g. ~/.config/gp-okta.conf') + parser.add_argument('--gpg-decrypt', action='store_true', + help='use gpg and settings from gpg-home to decrypt gpg encrypted conf_file') + parser.add_argument('--gpg-home', default=os.path.expanduser('~/.gnupg')) + parser.add_argument('--show-list-of-gateways', default=False, action='store_true', + help='get list of gateways from portal') + parser.add_argument('--quiet', default=False, action='store_true', + help='disable verbose logging') + args = parser.parse_args() + + global quiet + quiet = args.quiet + + assert os.path.exists(args.conf_file) + assert not args.gpg_decrypt or os.path.isdir(args.gpg_home) + + config_contents = '' + with io.open(args.conf_file, 'rb') as fp: + config_contents = fp.read() + + if args.conf_file.endswith('.gpg') and not args.gpg_decrypt: + err('conf file looks like gpg encrypted. Did you forget the --gpg-decrypt?') + + if args.gpg_decrypt: + if not have_gnupg: + err('Need gnupg package for reading gnupg encrypted files. Consider doing `pip install python-gnupg` (or similar)') + gpg = gnupg.GPG(gnupghome=args.gpg_home) + decrypted_contents = gpg.decrypt(config_contents) + + if not decrypted_contents.ok: + print('[ERROR] failed to decrypt config file:', file=sys.stderr) + print('[ERROR] status: {}'.format(decrypted_contents.status), file=sys.stderr) + print('[ERROR] error: {}'.format(decrypted_contents.stderr), file=sys.stderr) + sys.exit(1) + + config_contents = decrypted_contents.data + + conf = load_conf(config_contents) + s = requests.Session() s.headers['User-Agent'] = 'PAN GlobalProtect' - saml_xml = paloalto_prelogin(conf, s) + + if conf.get('client_cert'): + s.cert = conf.get('client_cert') + + if args.show_list_of_gateways: + userauthcookie, gateways = paloalto_getconfig(conf, s) + print("Gateways:") + for gateway in sorted(gateways): + print(" ", gateway) + sys.exit(0) + + another_dance = conf.get('another_dance', '').lower() in ['1', 'true'] + + if conf.get('gateway'): + gateway = conf.get('gateway').strip() + else: + gateway = None + + if gateway and not another_dance: + vpn_url = 'https://{0}'.format(gateway) + if vpn_url != conf.get('vpn_url'): + log('Discarding \'vpn_url\', as concrete gateway is given and another_dance = 0') + conf['vpn_url'] = vpn_url + + userauthcookie = None + + if another_dance or not gateway: + saml_xml = paloalto_prelogin(conf, s) + else: + saml_xml = paloalto_prelogin(conf, s, gateway) redirect_url = okta_saml(conf, s, saml_xml) token = okta_auth(conf, s) log('sessionToken: {0}'.format(token)) saml_username, prelogin_cookie = okta_redirect(conf, s, token, redirect_url) + if another_dance or not gateway: + userauthcookie, gateways = paloalto_getconfig(conf, s, saml_username, prelogin_cookie) + if not gateway and len(gateways): + gateway = gateways.pop() # this just grabs an arbitrary gateway + + log('portal-userauthcookie: {0}'.format(userauthcookie)) + log('gateway: {0}'.format(gateway)) log('saml-username: {0}'.format(saml_username)) log('prelogin-cookie: {0}'.format(prelogin_cookie)) - userauthcookie = paloalto_getconfig(conf, s, saml_username, prelogin_cookie) - log('portal-userauthcookie: {0}'.format(userauthcookie)) - + + if another_dance: + # 1st step: dance with the portal, 2nd step: dance with the gateway + saml_xml = paloalto_prelogin(conf, s, gateway) + saml_username, prelogin_cookie = okta_saml_2(conf, s, gateway, saml_xml) + log('saml-username (2): {0}'.format(saml_username)) + log('prelogin-cookie (2): {0}'.format(prelogin_cookie)) + + if (not userauthcookie or userauthcookie == 'empty') and prelogin_cookie != 'empty': + cookie_type = 'gateway:prelogin-cookie' + cookie = prelogin_cookie + else: + cookie_type = 'portal:portal-userauthcookie' + cookie = userauthcookie + username = saml_username - cmd = conf.get('openconnect_cmd') or 'openconnect' + + cmd = conf.get('openconnect_cmd', 'openconnect') cmd += ' --protocol=gp -u \'{0}\'' - cmd += ' --usergroup portal:portal-userauthcookie' - cmd += ' --passwd-on-stdin ' + conf.get('openconnect_args', '') + ' \'{1}\'' - cmd = cmd.format(username, conf.get('vpn_url')) - gw = (conf.get('gateway') or '').strip() + cmd += ' --usergroup {1}' + if conf.get('client_cert'): + cmd += ' --certificate=\'{0}\''.format(conf.get('client_cert')) + if conf.get('openconnect_certs') and os.path.getsize(conf.get('openconnect_certs').name) > 0: + cmd += ' --cafile=\'{0}\''.format(conf.get('openconnect_certs').name) + cmd += ' --passwd-on-stdin ' + conf.get('openconnect_args', '') + ' \'{2}\'' + cmd = cmd.format(username, cookie_type, + 'https://{0}'.format(gateway) if conf.get('another_dance', '').lower() in ['1', 'true'] else conf.get('vpn_url')) + bugs = '' if conf.get('bug.nl', '').lower() in ['1', 'true']: bugs += '\\n' if conf.get('bug.username', '').lower() in ['1', 'true']: bugs += '{0}\\n'.format(username.replace('\\', '\\\\')) - if len(gw) > 0: - pcmd = 'printf \'' + bugs + '{0}\\n{1}\''.format(userauthcookie, gw) + if len(gateway) > 0: + pcmd = 'printf \'' + bugs + '{0}\\n{1}\''.format(cookie, gateway) else: - pcmd = 'printf \'' + bugs + '{0}\''.format(userauthcookie) + pcmd = 'printf \'' + bugs + '{0}\''.format(cookie) print() if conf.get('execute', '').lower() in ['1', 'true']: cmd = shlex.split(cmd) @@ -458,7 +767,8 @@ def main(): cp.communicate() else: print('{0} | {1}'.format(pcmd, cmd)) + return 0 if __name__ == '__main__': - main() + sys.exit(main())