s0md3v/Corsy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #37 from Vasco-jofra/bug-fixes

Browse Source 
Fixes the "origin reflected" check and the extractHeaders function
This commit is contained in:
Somdev Sangwan
2021-11-20 10:35:49 +05:30
committed by GitHub
parent 06f4d4a06d 9f37125c04
commit 2985ae24da
3 changed files with 86 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
core/requester.py
View File

@@ -10,10 +10,11 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def requester(url, scheme, headers, origin): def requester(url, scheme, headers, origin):
headers['Origin'] = origin headers['Origin'] = origin
try: try:
response = requests.get(url, headers=headers, verify=False).headers response = requests.get(url, headers=headers, verify=False)
for key, value in response.items(): headers = response.headers
for key, value in headers.items():
if key.lower() == 'access-control-allow-origin': if key.lower() == 'access-control-allow-origin':
return response return headers
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
if 'Failed to establish a new connection' in str(e): if 'Failed to establish a new connection' in str(e):
print ('%s %s is unreachable' % (bad, url)) print ('%s %s is unreachable' % (bad, url))

144
core/tests.py
View File

@@ -25,85 +25,87 @@ def passive_tests(url, headers):
def active_tests(url, root, scheme, header_dict, delay): def active_tests(url, root, scheme, header_dict, delay):
origin = scheme + '://' + root origin = scheme + '://' + root
headers = requester(url, scheme, header_dict, origin) headers = requester(url, scheme, header_dict, origin)
if headers: acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
origin = root + '://' + 'example.com' if acao_header is None:
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None) return
if acao_header and acao_header == (origin):
info = details['origin reflected'] origin = scheme + '://' + 'example.com'
info['acao header'] = acao_header headers = requester(url, scheme, header_dict, origin)
info['acac header'] = acac_header acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
return {url : info} if acao_header and acao_header == (origin):
elif not acao_header: info = details['origin reflected']
return info['acao header'] = acao_header
time.sleep(delay) info['acac header'] = acac_header
return {url : info}
time.sleep(delay)
origin = scheme + '://' + root + '.example.com' origin = scheme + '://' + root + '.example.com'
headers = requester(url, scheme, header_dict, origin) headers = requester(url, scheme, header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None) acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header == (origin): if acao_header and acao_header == (origin):
info = details['post-domain wildcard'] info = details['post-domain wildcard']
info['acao header'] = acao_header info['acao header'] = acao_header
info['acac header'] = acac_header info['acac header'] = acac_header
return {url : info} return {url : info}
time.sleep(delay) time.sleep(delay)
origin = scheme + '://d3v' + root origin = scheme + '://d3v' + root
headers = requester(url, scheme, header_dict, origin) headers = requester(url, scheme, header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None) acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header == (origin): if acao_header and acao_header == (origin):
info = details['pre-domain wildcard'] info = details['pre-domain wildcard']
info['acao header'] = acao_header info['acao header'] = acao_header
info['acac header'] = acac_header info['acac header'] = acac_header
return {url : info} return {url : info}
time.sleep(delay) time.sleep(delay)
origin = 'null' origin = 'null'
headers = requester(url, '', header_dict, origin) headers = requester(url, '', header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None) acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header == 'null': if acao_header and acao_header == 'null':
info = details['null origin allowed'] info = details['null origin allowed']
info['acao header'] = acao_header info['acao header'] = acao_header
info['acac header'] = acac_header info['acac header'] = acac_header
return {url : info} return {url : info}
time.sleep(delay) time.sleep(delay)
origin = scheme + '://' + root + '_.example.com' origin = scheme + '://' + root + '_.example.com'
headers = requester(url, scheme, header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header == origin:
info = details['unrecognized underscore']
info['acao header'] = acao_header
info['acac header'] = acac_header
return {url : info}
time.sleep(delay)
origin = scheme + '://' + root + '%60.example.com'
headers = requester(url, scheme, header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and '`.example.com' in acao_header:
info = details['broken parser']
info['acao header'] = acao_header
info['acac header'] = acac_header
return {url : info}
time.sleep(delay)
if root.count('.') > 1:
origin = scheme + '://' + root.replace('.', 'x', 1)
headers = requester(url, scheme, header_dict, origin) headers = requester(url, scheme, header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None) acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header == origin: if acao_header and acao_header == origin:
info = details['unrecognized underscore'] info = details['unescaped regex']
info['acao header'] = acao_header info['acao header'] = acao_header
info['acac header'] = acac_header info['acac header'] = acac_header
return {url : info} return {url : info}
time.sleep(delay) time.sleep(delay)
origin = 'http://' + root
origin = scheme + '://' + root + '%60.example.com' headers = requester(url, 'http', header_dict, origin)
headers = requester(url, scheme, header_dict, origin) acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None) if acao_header and acao_header.startswith('http://'):
if acao_header and '`.example.com' in acao_header: info = details['http origin allowed']
info = details['broken parser'] info['acao header'] = acao_header
info['acao header'] = acao_header info['acac header'] = acac_header
info['acac header'] = acac_header return {url : info}
return {url : info} else:
time.sleep(delay) return passive_tests(url, headers)
if root.count('.') > 1:
origin = scheme + '://' + root.replace('.', 'x', 1)
headers = requester(url, scheme, header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header == origin:
info = details['unescaped regex']
info['acao header'] = acao_header
info['acac header'] = acac_header
return {url : info}
time.sleep(delay)
origin = 'http://' + root
headers = requester(url, 'http', header_dict, origin)
acao_header, acac_header = headers.get('access-control-allow-origin', None), headers.get('access-control-allow-credentials', None)
if acao_header and acao_header.startswith('http://'):
info = details['http origin allowed']
info['acao header'] = acao_header
info['acac header'] = acac_header
return {url : info}
else:
return passive_tests(url, headers)

22
core/utils.py
View File

@@ -64,17 +64,13 @@ def prompt(default=None):
return tmpfile.read().strip() return tmpfile.read().strip()
def extractHeaders(headers): def extractHeaders(headers: str):
headers = headers.replace('\\n', '\n')
sorted_headers = {} sorted_headers = {}
matches = re.findall(r'^?(.*?):\s(.*?)[\n$]', headers) for header in headers.split('\\n'):
for match in matches: name, value = header.split(":", 1)
header = match[0] name = name.strip()
value = match[1] value = value.strip()
try: if len(value) >= 1 and value[-1] == ',':
if value[-1] == ',': value = value[:-1]
value = value[:-1] sorted_headers[name] = value
sorted_headers[header] = value return sorted_headers
except IndexError:
pass
return sorted_headers