os-mnemo/pan-globalprotect-okta
1
0
Fork 0
mirror of https://github.com/os-mnemo/pan-globalprotect-okta synced 2026-07-31 07:44:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #19 from coldcoff/master

Browse Source 
Several improvements for `gp-okta.py`
This commit is contained in:
Andris Raugulis
2020-05-21 13:18:32 +03:00
committed by GitHub
parent a808b2486d 7fbaeee604
commit ba14b46d7d
2 changed files with 420 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gp-okta.conf
+15 -3
View File
@@ -1,15 +1,27 @@
debug = 0 debug = 0
vpn_url = https://vpn.example.com vpn_url = https://vpn.example.com
#vpn_url_cert = vpn_url.cert
okta_url = https://example.okta.com okta_url = https://example.okta.com
#okta_url_cert = okta_url.cert
username = myuser username = myuser
password = mypass password = mypass
#client_cert = path-to-myusers-client-cert-as-unencrypted-pem-file.pem
# mfa_order = totp sms # mfa_order = totp sms
sms.okta = 0 sms.okta = 0
totp.okta = ABCDEFGHIJKLMNOP #totp.okta = ABCDEFGHIJKLMNOP
totp.google = ABCDEFGHIJKLMNOP #totp.google = ABCDEFGHIJKLMNOP
gateway = Manual ny1-gw.example.com #totp.symantec = ABCDEFGHIJKLMNOP
gateway = Manual ny1-gw.example.com # optional hardcoded gateway
#openconnect_cmd = sudo openconnect #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 openconnect_args = # optional arguments to openconnect
execute = 0 # execute openconnect command execute = 0 # execute openconnect command
another_dance = 0 # second round of authentication required
bug.nl = 0 # newline work-around for openconnect bug.nl = 0 # newline work-around for openconnect
bug.username = 0 # username work-around for openconnect bug.username = 0 # username work-around for openconnect
gp-okta.py
+405 -95
View File
@@ -2,19 +2,24 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
The MIT License (MIT) The MIT License (MIT)
Copyright (C) 2018 Andris Raugulis (moo@arthepsy.eu) 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 Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions: furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software. all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 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 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE. THE SOFTWARE.
""" """
from __future__ import print_function from __future__ import print_function, unicode_literals
import io, os, sys, re, json, base64, getpass, subprocess, shlex, signal import io, os, sys, time, re, json, base64, getpass, subprocess, shlex, signal, tempfile, traceback
from lxml import etree from lxml import etree
import requests import requests
import argparse
if sys.version_info >= (3,): if sys.version_info >= (3,):
from urllib.parse import urljoin from urllib.parse import urlparse, urljoin
text_type = str text_type = str
binary_type = bytes binary_type = bytes
else: else:
from urlparse import urljoin from urlparse import urlparse, urljoin
text_type = unicode text_type = unicode
binary_type = str binary_type = str
input = raw_input 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_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') to_u = lambda v: v if isinstance(v, text_type) else v.decode('utf-8')
quiet = False
def log(s): def log(s):
print('[INFO] {0}'.format(s)) if not quiet:
print('[INFO] {0}'.format(s))
def dbg(d, h, *xs): def dbg(d, h, *xs):
if quiet:
return
if not d: if not d:
return return
print('# {0}:'.format(h)) print('[DEBUG] {0}:'.format(h))
for x in xs: for x in xs:
print(x) print('[DEBUG] {0}'.format(x))
print('---') print('[DEBUG] ---')
def err(s): def err(s):
print('err: {0}'.format(s)) print('[ERROR] {0}'.format(s), file=sys.stderr)
sys.exit(1) sys.exit(1)
def parse_xml(xml): def parse_xml(xml):
@@ -62,21 +99,21 @@ def parse_xml(xml):
xml = bytes(bytearray(xml, encoding='utf-8')) xml = bytes(bytearray(xml, encoding='utf-8'))
parser = etree.XMLParser(ns_clean=True, recover=True) parser = etree.XMLParser(ns_clean=True, recover=True)
return etree.fromstring(xml, parser) return etree.fromstring(xml, parser)
except: except Exception as e:
err('failed to parse xml') err('failed to parse xml: ' + e)
def parse_html(html): def parse_html(html):
try: try:
parser = etree.HTMLParser() parser = etree.HTMLParser()
return etree.fromstring(html, parser) return etree.fromstring(html, parser)
except: except Exception as e:
err('failed to parse html') err('failed to parse html: ' + e)
def parse_rjson(r): def parse_rjson(r):
try: try:
return r.json() return r.json()
except: except Exception as e:
err('failed to parse json') err('failed to parse json: ' + e)
def parse_form(html, current_url = None): def parse_form(html, current_url = None):
xform = html.find('.//form') xform = html.find('.//form')
@@ -91,25 +128,26 @@ def parse_form(html, current_url = None):
data[k] = v data[k] = v
return url, data return url, data
def load_conf(cf): def load_conf(cf):
log('load conf')
conf = {} conf = {}
keys = ['vpn_url', 'username', 'password', 'okta_url'] keys = ['vpn_url', 'username', 'password', 'okta_url']
if isinstance(cf, binary_type):
cf = cf.decode('utf-8')
line_nr = 0 line_nr = 0
with io.open(cf, 'r', encoding='utf-8') as fp: for rline in cf.split('\n'):
for rline in fp: line_nr += 1
line_nr += 1 line = rline.strip()
line = rline.strip() mx = re.match(r'^\s*([^=\s]+)\s*=\s*(.*?)\s*(?:#\s+.*)?\s*$', line)
mx = re.match(r'^\s*([^=\s]+)\s*=\s*(.*?)\s*(?:#\s+.*)?\s*$', line) if mx:
if mx: k, v = mx.group(1).lower(), mx.group(2)
k, v = mx.group(1).lower(), mx.group(2) if k.startswith('#'):
if k.startswith('#'): continue
continue for q in '"\'':
for q in '"\'': if re.match(r'^{0}.*{0}$'.format(q), v):
if re.match(r'^{0}.*{0}$'.format(q), v): v = v[1:-1]
v = v[1:-1] conf[k] = v
conf[k] = v conf['{0}.line'.format(k)] = line_nr
conf['{0}.line'.format(k)] = line_nr
for k, v in os.environ.items(): for k, v in os.environ.items():
k = k.lower() k = k.lower()
if k.startswith('gp_'): if k.startswith('gp_'):
@@ -121,6 +159,22 @@ def load_conf(cf):
conf['username'] = input('username: ').strip() conf['username'] = input('username: ').strip()
if len(conf.get('password', '').strip()) == 0: if len(conf.get('password', '').strip()) == 0:
conf['password'] = getpass.getpass('password: ').strip() 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: for k in keys:
if k not in conf: if k not in conf:
err('missing configuration key: {0}'.format(k)) err('missing configuration key: {0}'.format(k))
@@ -128,12 +182,13 @@ def load_conf(cf):
if len(conf[k].strip()) == 0: if len(conf[k].strip()) == 0:
err('empty configuration key: {0}'.format(k)) err('empty configuration key: {0}'.format(k))
conf['debug'] = conf.get('debug', '').lower() in ['1', 'true'] conf['debug'] = conf.get('debug', '').lower() in ['1', 'true']
conf['openconnect_certs'].flush()
return conf return conf
def mfa_priority(conf, ftype, fprovider): def mfa_priority(conf, ftype, fprovider):
if ftype == 'token:software:totp': if ftype == 'token:software:totp' or (ftype, fprovider) == ('token', 'symantec'):
ftype = 'totp' ftype = 'totp'
if ftype not in ['totp', 'sms']: if ftype not in ['totp', 'sms', 'push', 'webauthn']:
return 0 return 0
mfa_order = conf.get('mfa_order', '') mfa_order = conf.get('mfa_order', '')
if ftype in mfa_order: if ftype in mfa_order:
@@ -141,7 +196,7 @@ def mfa_priority(conf, ftype, fprovider):
else: else:
priority = 0 priority = 0
value = conf.get('{0}.{1}'.format(ftype, fprovider)) value = conf.get('{0}.{1}'.format(ftype, fprovider))
if ftype == 'sms': if ftype in ('sms', 'webauthn'):
if not (value or '').lower() in ['1', 'true']: if not (value or '').lower() in ['1', 'true']:
value = None value = None
line_nr = conf.get('{0}.{1}.line'.format(ftype, fprovider), 0) 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) priority += (512 - line_nr)
return priority 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): def get_redirect_url(conf, c, current_url = None):
rx_base_url = re.search(r'var\s*baseUrl\s*=\s*\'([^\']+)\'', c) 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): def send_req(conf, s, name, url, data, **kwargs):
dbg(conf.get('debug'), '{0}.request'.format(name), url) 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 do_json = True if kwargs.get('json') else False
headers = {} headers = {}
if do_json: if do_json:
@@ -180,48 +249,67 @@ def send_req(conf, s, name, url, data, **kwargs):
headers['Accept'] = 'application/json' headers['Accept'] = 'application/json'
headers['Content-Type'] = 'application/json' headers['Content-Type'] = 'application/json'
if kwargs.get('get'): if kwargs.get('get'):
r = s.get(url, headers=headers) r = s.get(url, headers=headers,
verify=kwargs.get('verify', True))
else: 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())]) 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) rr = 'status: {0}\n\n{1}\n\n{2}'.format(r.status_code, hdump, r.text)
if r.status_code != 200: 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) dbg(conf.get('debug'), '{0}.response'.format(name), rr)
if do_json: if do_json:
return r.headers, parse_rjson(r) return r.headers, parse_rjson(r)
return r.headers, r.text return r.headers, r.text
def paloalto_prelogin(conf, s, gateway=None):
def paloalto_prelogin(conf, s): verify = None
log('prelogin request') if gateway:
url = '{0}/global-protect/prelogin.esp'.format(conf.get('vpn_url')) # 2nd round: use gateway
h, c = send_req(conf, s, 'prelogin', url, {}, get=True) 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) x = parse_xml(c)
saml_req = x.find('.//saml-request') saml_req = x.find('.//saml-request')
if saml_req is None: 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: if len(saml_req.text.strip()) == 0:
err('empty saml request') err('empty saml request')
try: try:
saml_raw = base64.b64decode(saml_req.text) saml_raw = base64.b64decode(saml_req.text)
except: except Exception as e:
err('failed to decode saml request') err('failed to decode saml request: ' + e)
dbg(conf.get('debug'), 'prelogin.decoded', saml_raw) dbg(conf.get('debug'), 'prelogin.decoded', saml_raw)
saml_xml = parse_html(saml_raw) saml_xml = parse_html(saml_raw)
return saml_xml return saml_xml
def okta_saml(conf, s, 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) 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) redirect_url = get_redirect_url(conf, c, url)
if redirect_url is None: if redirect_url is None:
err('did not find redirect url') err('did not find redirect url')
return redirect_url return redirect_url
def okta_auth(conf, s): def okta_auth(conf, s, stateToken = None):
log('okta auth request') log('okta auth request [okta_url]')
url = '{0}/api/v1/authn'.format(conf.get('okta_url')) url = '{0}/api/v1/authn'.format(conf.get('okta_url'))
data = { data = {
'username': conf.get('username'), 'username': conf.get('username'),
@@ -230,8 +318,11 @@ def okta_auth(conf, s):
'warnBeforePasswordExpired':True, 'warnBeforePasswordExpired':True,
'multiOptionalFactorEnroll':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: while True:
ok, r = okta_transaction_state(conf, s, j) ok, r = okta_transaction_state(conf, s, j)
@@ -254,7 +345,8 @@ def okta_transaction_state(conf, s, j):
if len(state_token) == 0: if len(state_token) == 0:
err('empty state token') err('empty state token')
data = {'stateToken': 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 return False, j
# status: password_expired # status: password_expired
# status: recovery # status: recovery
@@ -302,12 +394,17 @@ def okta_mfa(conf, s, j):
if len(factors) == 0: if len(factors) == 0:
err('no factors found') err('no factors found')
for f in sorted(factors, key=lambda x: x.get('priority', 0), reverse=True): for f in sorted(factors, key=lambda x: x.get('priority', 0), reverse=True):
print(f) #print(f)
ftype = f.get('type') 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) r = okta_mfa_totp(conf, s, f, state_token)
elif ftype == 'sms': elif ftype == 'sms':
r = okta_mfa_sms(conf, s, f, state_token) 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: else:
r = None r = None
if r is not None: if r is not None:
@@ -321,7 +418,8 @@ def okta_mfa_totp(conf, s, factor, state_token):
if len(secret) == 0: if len(secret) == 0:
code = input('{0} TOTP: '.format(provider)).strip() code = input('{0} TOTP: '.format(provider)).strip()
else: else:
import pyotp if not have_pyotp:
err('Need pyotp package, consider doing \'pip install pyotp\' (or similar)')
totp = pyotp.TOTP(secret) totp = pyotp.TOTP(secret)
code = totp.now() code = totp.now()
code = code or '' code = code or ''
@@ -332,23 +430,93 @@ def okta_mfa_totp(conf, s, factor, state_token):
'stateToken': state_token, 'stateToken': state_token,
'passCode': code 'passCode': code
} }
log('mfa {0} totp request'.format(provider)) 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) _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 return j
def okta_mfa_sms(conf, s, factor, state_token): def okta_mfa_sms(conf, s, factor, state_token):
provider = factor.get('provider', '') provider = factor.get('provider', '')
data = { data = {
'factorId': factor.get('id'), 'factorId': factor.get('id'),
'stateToken': state_token, 'stateToken': state_token
} }
log('mfa {0} sms request'.format(provider)) log('mfa {0} sms request [okta_url]'.format(provider))
h, j = send_req(conf, s, 'sms mfa', factor.get('url'), data, json=True) _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() code = input('{0} SMS verification code: '.format(provider)).strip()
if len(code) == 0: if len(code) == 0:
return None return None
data['passCode'] = code 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 return j
def okta_redirect(conf, s, session_token, redirect_url): 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 'redirectUrl': redirect_url
} }
url = '{0}/login/sessionCookieRedirect'.format(conf.get('okta_url')) url = '{0}/login/sessionCookieRedirect'.format(conf.get('okta_url'))
log('okta redirect request') log('okta redirect request {0} [okta_url]'.format(rc))
h, c = send_req(conf, s, 'redirect', url, data) 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) redirect_url = get_redirect_url(conf, c, url)
if redirect_url: if redirect_url:
form_url, form_data = None, {} form_url, form_data = None, {}
else: else:
xhtml = parse_html(c) xhtml = parse_html(c)
form_url, form_data = parse_form(xhtml, url) 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: elif form_url:
log('okta redirect form request') log('okta redirect form request [vpn_url]')
h, c = send_req(conf, s, 'redirect form', form_url, form_data) 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() saml_username = h.get('saml-username', '').strip()
prelogin_cookie = h.get('prelogin-cookie', '').strip() prelogin_cookie = h.get('prelogin-cookie', '').strip()
if saml_username and prelogin_cookie: 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]) dbg(conf.get('debug'), 'saml prop', [saml_auth_status, saml_slo])
return saml_username, prelogin_cookie return saml_username, prelogin_cookie
def paloalto_getconfig(conf, s, saml_username, prelogin_cookie): def paloalto_getconfig(conf, s, username = None, prelogin_cookie = None):
log('getconfig request') log('getconfig request [vpn_url]')
url = '{0}/global-protect/getconfig.esp'.format(conf.get('vpn_url')) url = '{0}/global-protect/getconfig.esp'.format(conf.get('vpn_url'))
data = { data = {
'user': saml_username, #'jnlpReady': 'jnlpReady',
'passwd': '', #'ok': 'Login',
'inputStr': '', #'direct': 'yes',
'clientVer': '4100', 'clientVer': '4100',
#'prot': 'https:',
'clientos': 'Windows', 'clientos': 'Windows',
'clientgpversion': '4.1.0.98',
'computer': 'DESKTOP',
'os-version': 'Microsoft Windows 10 Pro, 64-bit', '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' # 'host-id': '00:11:22:33:44:55'
'prelogin-cookie': prelogin_cookie, 'prelogin-cookie': prelogin_cookie or '',
'ipv6-support': 'yes' '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) x = parse_xml(c)
xtmp = x.find('.//portal-userauthcookie') xtmp = x.find('.//portal-userauthcookie')
if xtmp is None: if xtmp is None:
@@ -409,43 +590,171 @@ def paloalto_getconfig(conf, s, saml_username, prelogin_cookie):
portal_userauthcookie = xtmp.text portal_userauthcookie = xtmp.text
if len(portal_userauthcookie) == 0: if len(portal_userauthcookie) == 0:
err('empty portal_userauthcookie') 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(): def main():
if len(sys.argv) < 2:
print('usage: {0} <conf>'.format(sys.argv[0])) parser = argparse.ArgumentParser(description="""
sys.exit(1) This is an OpenConnect wrapper script that automates connecting to a
conf = load_conf(sys.argv[1]) 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 = requests.Session()
s.headers['User-Agent'] = 'PAN GlobalProtect' 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) redirect_url = okta_saml(conf, s, saml_xml)
token = okta_auth(conf, s) token = okta_auth(conf, s)
log('sessionToken: {0}'.format(token)) log('sessionToken: {0}'.format(token))
saml_username, prelogin_cookie = okta_redirect(conf, s, token, redirect_url) 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('saml-username: {0}'.format(saml_username))
log('prelogin-cookie: {0}'.format(prelogin_cookie)) 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 username = saml_username
cmd = conf.get('openconnect_cmd') or 'openconnect'
cmd = conf.get('openconnect_cmd', 'openconnect')
cmd += ' --protocol=gp -u \'{0}\'' cmd += ' --protocol=gp -u \'{0}\''
cmd += ' --usergroup portal:portal-userauthcookie' cmd += ' --usergroup {1}'
cmd += ' --passwd-on-stdin ' + conf.get('openconnect_args', '') + ' \'{1}\'' if conf.get('client_cert'):
cmd = cmd.format(username, conf.get('vpn_url')) cmd += ' --certificate=\'{0}\''.format(conf.get('client_cert'))
gw = (conf.get('gateway') or '').strip() 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 = '' bugs = ''
if conf.get('bug.nl', '').lower() in ['1', 'true']: if conf.get('bug.nl', '').lower() in ['1', 'true']:
bugs += '\\n' bugs += '\\n'
if conf.get('bug.username', '').lower() in ['1', 'true']: if conf.get('bug.username', '').lower() in ['1', 'true']:
bugs += '{0}\\n'.format(username.replace('\\', '\\\\')) bugs += '{0}\\n'.format(username.replace('\\', '\\\\'))
if len(gw) > 0: if len(gateway) > 0:
pcmd = 'printf \'' + bugs + '{0}\\n{1}\''.format(userauthcookie, gw) pcmd = 'printf \'' + bugs + '{0}\\n{1}\''.format(cookie, gateway)
else: else:
pcmd = 'printf \'' + bugs + '{0}\''.format(userauthcookie) pcmd = 'printf \'' + bugs + '{0}\''.format(cookie)
print() print()
if conf.get('execute', '').lower() in ['1', 'true']: if conf.get('execute', '').lower() in ['1', 'true']:
cmd = shlex.split(cmd) cmd = shlex.split(cmd)
@@ -458,7 +767,8 @@ def main():
cp.communicate() cp.communicate()
else: else:
print('{0} | {1}'.format(pcmd, cmd)) print('{0} | {1}'.format(pcmd, cmd))
return 0
if __name__ == '__main__': if __name__ == '__main__':
main() sys.exit(main())