SSLv2 Drown Attack Scan

#!/usr/bin/env python

import sys
from enum import Enum
import time
import datetime
import socket
import Crypto.Cipher
import signal
from binascii import hexlify
import base64

#!/usr/bin/env python

import sys
from enum import Enum
import time
import datetime
import socket
import Crypto.Cipher
import signal
from binascii import hexlify
import base64

sys.path.append("./scapy-ssl_tls/")
import logging
logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
import scapy
from scapy.all import *
from ssl_tls import *
import ssl_tls_crypto

from pyx509.pkcs7.asn1_models.X509_certificate import Certificate
from pyx509.pkcs7_models import X509Certificate, PublicKeyInfo, ExtendedKeyUsageExt
from pyx509.pkcs7.asn1_models.decoder_workarounds import decode

import select

SOCKET_TIMEOUT = 15
SOCKET_RECV_SIZE = 80 * 1024

CON_FAIL = "con fail"
NO_STARTTLS = "no starttls"
NO_TLS = "no tls"
VULN = "vuln"

def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
    import signal

    class TimeoutError(Exception):
        pass

    def handler(signum, frame):
        raise TimeoutError()

    # set the timeout handler
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(timeout_duration)
    try:
        result = func(*args, **kwargs)
    except TimeoutError as exc:
        result = default
    finally:
        signal.alarm(0)

    return result

CHALLENGE    = 'a' * 16
CLEAR_KEY    = '\0' * 11
KEY_ARGUMENT = '\0' * 8

class CipherSuite(object):
    @classmethod
    def get_string_description(cls):
        raise NotImplementedError()

    @classmethod
    def get_constant(cls):
        return eval("SSLv2CipherSuite." + cls.get_string_description())

    @classmethod
    def get_client_master_key(cls, encrypted_pms):
        raise NotImplementedError()

    @classmethod
    def verify_key(cls, connection_id, server_finished):
        raise NotImplementedError()

    @classmethod
    def get_encrypted_pms(cls, public_key, secret_key):
        pkcs1_pubkey = Crypto.Cipher.PKCS1_v1_5.new(public_key)
        encrypted_pms = pkcs1_pubkey.encrypt(secret_key)
        return encrypted_pms

class RC4Export(CipherSuite):
    SECRET_KEY = 'b' * 5

    @classmethod
    def get_string_description(cls):
        return "RC4_128_EXPORT40_WITH_MD5"

    @classmethod
    def get_client_master_key(cls, public_key):
        client_master_key = SSLv2ClientMasterKey(cipher_suite=cls.get_constant(),
                                                 encrypted_key=cls.get_encrypted_pms(public_key, cls.SECRET_KEY),
                                                 clear_key=CLEAR_KEY)
        return client_master_key

    @classmethod
    def verify_key(cls, connection_id, server_finished):
        md5 = MD5.new(CLEAR_KEY + cls.SECRET_KEY + '0' + CHALLENGE + connection_id).digest()
        rc4 = Crypto.Cipher.ARC4.new(md5)
        if not rc4.decrypt(server_finished[2:]).endswith(CHALLENGE):
            return False
        return True

class RC4(CipherSuite):
    SECRET_KEY = 'b' * 16
    CLEAR_KEY  = 'a' * 15

    @classmethod
    def get_string_description(cls):
        return "RC4_128_WITH_MD5"

    @classmethod
    def get_client_master_key(cls, public_key):
        client_master_key = SSLv2ClientMasterKey(cipher_suite=cls.get_constant(),
                                                 encrypted_key=cls.get_encrypted_pms(public_key, cls.SECRET_KEY),
                                                 clear_key=cls.CLEAR_KEY)
        return client_master_key

    @classmethod
    def verify_key(cls, connection_id, server_finished):
        md5 = MD5.new((cls.CLEAR_KEY + cls.SECRET_KEY)[:16] + '0' + CHALLENGE + connection_id).digest()
        rc4 = Crypto.Cipher.ARC4.new(md5)
        if not rc4.decrypt(server_finished[2:]).endswith(CHALLENGE):
            return False
        return True

class RC2Export(CipherSuite):
    SECRET_KEY = 'b' * 5

    @classmethod
    def get_string_description(cls):
        return "RC2_128_CBC_EXPORT40_WITH_MD5"

    @classmethod
    def get_client_master_key(cls, public_key):
        client_master_key = SSLv2ClientMasterKey(cipher_suite=cls.get_constant(),
                                                 encrypted_key=cls.get_encrypted_pms(public_key, cls.SECRET_KEY),
                                                 key_argument=KEY_ARGUMENT,
                                                 clear_key=CLEAR_KEY)
        return client_master_key

    @classmethod
    def verify_key(cls, connection_id, server_finished):
        md5 = MD5.new(CLEAR_KEY + cls.SECRET_KEY + '0' + CHALLENGE + connection_id).digest()
        rc2 = Crypto.Cipher.ARC2.new(md5, mode=Crypto.Cipher.ARC2.MODE_CBC, IV=KEY_ARGUMENT, effective_keylen=128)
        try:
            decryption = rc2.decrypt(server_finished[3:])
        except ValueError, e:
            return False
        if decryption[17:].startswith(CHALLENGE):
            return True
        return False

class DES(CipherSuite):
    SECRET_KEY = 'b' * 8

    @classmethod
    def get_string_description(cls):
        return "DES_64_CBC_WITH_MD5"

    @classmethod
    def get_client_master_key(cls, public_key):
        client_master_key = SSLv2ClientMasterKey(cipher_suite=cls.get_constant(),
                                                 encrypted_key=cls.get_encrypted_pms(public_key, cls.SECRET_KEY),
                                                 key_argument=KEY_ARGUMENT)
        return client_master_key

    @classmethod
    def verify_key(cls, connection_id, server_finished):
        md5 = MD5.new(cls.SECRET_KEY + '0' + CHALLENGE + connection_id).digest()
        des = Crypto.Cipher.DES.new(md5[:8], mode=Crypto.Cipher.DES.MODE_CBC, IV=KEY_ARGUMENT)
        try:
            decryption = des.decrypt(server_finished[3:])
        except ValueError, e:
            return False
        if decryption[17:].startswith(CHALLENGE):
            return True
        return False

cipher_suites = [RC2Export, RC4Export, RC4, DES]

def parse_certificate(derData):
    cert = decode(derData, asn1Spec=Certificate())[0]
    x509cert = X509Certificate(cert)
    tbs = x509cert.tbsCertificate

    algType = tbs.pub_key_info.algType
    algParams = tbs.pub_key_info.key

    if (algType != PublicKeyInfo.RSA):
        print 'Certificate algType is not RSA'
        raise Exception()

    return RSA.construct((long(hexlify(algParams["mod"]), 16), long(algParams["exp"])))

class Protocol(Enum):
    BARE_SSLv2 = 1
    ESMTP      = 2
    IMAP       = 3
    POP3       = 4

def sslv2_connect(ip, port, protocol, cipher_suite, result_additional_data):
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    s.settimeout(SOCKET_TIMEOUT)
    try:
        s.connect((ip, port))
    except socket.error, e:
        try:
            s.connect((ip, port))
        except socket.error, e:
            try:
                s.connect((ip, port))
            except socket.error, e:
                print '%s: Case 1 - port is apparently closed (after 3 tries); Connect failed' % ip
                return CON_FAIL

    starttls_response = "n/a"
    try:
        if protocol == Protocol.ESMTP:
            banner = s.recv(SOCKET_RECV_SIZE)
            s.send("EHLO testing\r\n")
            ehlo_response = s.recv(SOCKET_RECV_SIZE)
            if "starttls" not in ehlo_response.lower():
                print "%s: Case 2a; Server apparently doesn't support STARTTLS" % ip
                return NO_STARTTLS
            s.send("STARTTLS\r\n")
            starttls_response = s.recv(SOCKET_RECV_SIZE)
        if protocol == Protocol.IMAP:
            banner = s.recv(SOCKET_RECV_SIZE)
            s.send(". CAPABILITY\r\n")
            banner = s.recv(SOCKET_RECV_SIZE)
            if "starttls" not in banner.lower():
                print "%s: Case 2b; Server apparently doesn't support STARTTLS" % ip
                return NO_STARTTLS
            s.send(". STARTTLS\r\n")
            starttls_response = s.recv(SOCKET_RECV_SIZE)
        if protocol == Protocol.POP3:
            banner = s.recv(SOCKET_RECV_SIZE)
            s.send("STLS\r\n")
            starttls_response = s.recv(SOCKET_RECV_SIZE)
    except socket.error, e:
        print "Errorx: " + str(e) + " - starttls_response: '" + starttls_response + "'"
        print '%s: Case 2c; starttls negotiation failed' % ip
        return NO_STARTTLS


    client_hello = SSLv2Record()/SSLv2ClientHello(cipher_suites=SSL2_CIPHER_SUITES.keys(),challenge=CHALLENGE)
    s.sendall(str(client_hello))

    rlist, wlist, xlist = select.select([s], [], [s], SOCKET_TIMEOUT)
    if s in xlist or not s in rlist:
        print '%s: Case 3a; Server did not response properly to client hello' % ip
        s.close()
        return "3a: %s" % NO_TLS

    try:
        server_hello_raw = s.recv(SOCKET_RECV_SIZE)
    except socket.error, e:
        print '%s: Case 3b; Connection reset by peer when waiting for server hello' % ip
        s.close()
        return "3b: %s" % NO_TLS

    server_hello = timeout(SSL, (server_hello_raw,), timeout_duration=SOCKET_TIMEOUT)
    if server_hello == None:
        print '%s: Case 3c; Timeout on parsing server hello' % ip
        s.close()
        return "3c: %s" % NO_TLS

    if not SSLv2ServerHello in server_hello:
        print '%s: Case 3d; Server hello did not contain server hello' % ip
        s.close()
        return "3d: %s" % NO_TLS

    parsed_server_hello = server_hello[SSLv2ServerHello]
    connection_id = parsed_server_hello.connection_id
    cert = parsed_server_hello.certificates

    try:
        public_key = parse_certificate(cert)
    except:
        # Server could still be vulnerable, we just can't tell, so we assume it isn't
        print '%s: Case 4a; Could not extract public key from DER' % ip
        s.close()
        return "4a: %s" % NO_TLS

    server_advertised_cipher_suites = parsed_server_hello.fields["cipher_suites"]
    cipher_suite_advertised = cipher_suite.get_constant() in server_advertised_cipher_suites

    client_master_key = cipher_suite.get_client_master_key(public_key)
    client_key_exchange = SSLv2Record()/client_master_key

    s.sendall(str(client_key_exchange))

    rlist, wlist, xlist = select.select([s], [], [s], SOCKET_TIMEOUT)
    if s in xlist:
        print '%s: Case 4b; Exception on socket after sending client key exchange' % ip
        s.close()
        return "4b: %s" % NO_TLS
    if not s in rlist:
        print '%s: Case 5; Server did not send finished in time' % ip
        s.close()
        return "5: %s" % NO_TLS
    try:
        server_finished = s.recv(SOCKET_RECV_SIZE)
    except socket.error, e:
        print '%s: Case 4c; Connection reset by peer when waiting for server finished' % ip
        s.close()
        return "4c: %s" % NO_TLS

    if server_finished == '':
        print '%s: Case 4d; Empty server_finished' % ip
        s.close()
        return "4d: %s" % NO_TLS

    if not cipher_suite.verify_key(connection_id, server_finished):
        print '%s: Case 7; Symmetric key did not successfully verify on server finished message' % ip
        return "7: %s" % NO_TLS

    s.close()

    result_additional_data['cipher_suite_advertised'] = cipher_suite_advertised
    return "%s:%s" % (VULN, base64.b64encode(public_key.exportKey(format='DER')))

if __name__ == '__main__':
    ip = sys.argv[1]
    port = int(sys.argv[2])
    scan_id = os.getcwd()
    dtime = datetime.datetime.now()
    print 'Testing %s on port %s' % (ip, port)

    protocol = Protocol.BARE_SSLv2
    if len(sys.argv) >= 4:
        if sys.argv[3] == '-esmtp':
            protocol = Protocol.ESMTP
        elif sys.argv[3] == '-imap':
            protocol = Protocol.IMAP
        elif sys.argv[3] == '-pop3':
            protocol = Protocol.POP3
        elif sys.argv[3] == '-bare':
            protocol = Protocol.BARE_SSLv2
        else:
            print 'You gave 3 arguments, argument 3 is not a recognized protocol. Bailing out'
            sys.exit(1)

    vulns = []
    for cipher_suite in cipher_suites:

        string_description = cipher_suite.get_string_description()
        ret_additional_data = {}
        ret = sslv2_connect(ip, port, protocol, cipher_suite, ret_additional_data)

        if ret.startswith(VULN):
            pub_key = ret.replace('%s:' % VULN, '')

            cve_string = ""
            if not ret_additional_data['cipher_suite_advertised']:
                cve_string = " to CVE-2015-3197"
            if string_description == "RC4_128_WITH_MD5":
                if cve_string == "":
                    cve_string = " to CVE-2016-0703"
                else:
                    cve_string += " and CVE-2016-0703"

            print '%s: Server is vulnerable%s, with cipher %s\n' % (ip, cve_string, string_description)
        else:
            print '%s: Server is NOT vulnerable with cipher %s, Message: %s\n' % (ip, string_description, ret)

• Tweet
• Share
• Share
• Share
• Share