#! python
import sys
import os
import os.path
import socket
import thread
import time
import httplib
import BaseHTTPServer
import SimpleHTTPServer
try:
from cryptoIDlib.api import *
cryptoIDlibLoaded = True
except:
cryptoIDlibLoaded = False
if __name__ != "__main__":
raise "This must be run as a command, not used as a module!"
#import tlslite
#from tlslite.constants import AlertDescription, Fault
#from tlslite.utils.jython_compat import formatExceptionTrace
#from tlslite.X509 import X509, X509CertChain
from tlslite.api import *
def parsePrivateKey(s):
try:
return parsePEMKey(s, private=True)
except Exception, e:
print e
return parseXMLKey(s, private=True)
def clientTest(address, dir):
#Split address into hostname/port tuple
address = address.split(":")
if len(address)==1:
address.append("4443")
address = ( address[0], int(address[1]) )
def connect():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if hasattr(sock, 'settimeout'): #It's a python 2.3 feature
sock.settimeout(5)
sock.connect(address)
c = TLSConnection(sock)
return c
test = 0
badFault = False
print "Test 1 - good shared key"
connection = connect()
connection.handshakeClientSharedKey("shared", "key")
connection.close()
connection.sock.close()
print "Test 2 - shared key faults"
for fault in Fault.clientSharedKeyFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientSharedKey("shared", "key")
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
print "Test 3 - good SRP"
connection = connect()
connection.handshakeClientSRP("test", "password")
connection.close()
print "Test 4 - SRP faults"
for fault in Fault.clientSrpFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientSRP("test", "password")
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
print "Test 5 - good SRP: unknown_srp_username idiom"
def srpCallback():
return ("test", "password")
connection = connect()
connection.handshakeClientUnknown(srpCallback=srpCallback)
connection.close()
connection.sock.close()
print "Test 6 - good SRP: with X.509 certificate"
connection = connect()
connection.handshakeClientSRP("test", "password")
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 7 - X.509 with SRP faults"
for fault in Fault.clientSrpFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientSRP("test", "password")
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
if cryptoIDlibLoaded:
print "Test 8 - good SRP: with cryptoID certificate chain"
connection = connect()
connection.handshakeClientSRP("test", "password")
assert(isinstance(connection.session.serverCertChain, CertChain))
if not (connection.session.serverCertChain.validate()):
print connection.session.serverCertChain.validate(listProblems=True)
connection.close()
connection.sock.close()
print "Test 9 - CryptoID with SRP faults"
for fault in Fault.clientSrpFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientSRP("test", "password")
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
print "Test 10 - good X509"
connection = connect()
connection.handshakeClientCert()
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 10.a - good X509, SSLv3"
connection = connect()
settings = HandshakeSettings()
settings.minVersion = (3,0)
settings.maxVersion = (3,0)
connection.handshakeClientCert(settings=settings)
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 11 - X.509 faults"
for fault in Fault.clientNoAuthFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientCert()
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
if cryptoIDlibLoaded:
print "Test 12 - good cryptoID"
connection = connect()
connection.handshakeClientCert()
assert(isinstance(connection.session.serverCertChain, CertChain))
assert(connection.session.serverCertChain.validate())
connection.close()
connection.sock.close()
print "Test 13 - cryptoID faults"
for fault in Fault.clientNoAuthFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientCert()
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
print "Test 14 - good mutual X509"
x509Cert = X509().parse(open(os.path.join(dir, "clientX509Cert.pem")).read())
x509Chain = X509CertChain([x509Cert])
s = open(os.path.join(dir, "clientX509Key.pem")).read()
x509Key = parsePEMKey(s, private=True)
connection = connect()
connection.handshakeClientCert(x509Chain, x509Key)
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 14.a - good mutual X509, SSLv3"
connection = connect()
settings = HandshakeSettings()
settings.minVersion = (3,0)
settings.maxVersion = (3,0)
connection.handshakeClientCert(x509Chain, x509Key, settings=settings)
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 15 - mutual X.509 faults"
for fault in Fault.clientCertFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientCert(x509Chain, x509Key)
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
if cryptoIDlibLoaded:
print "Test 16 - good mutual cryptoID"
cryptoIDChain = CertChain().parse(open(os.path.join(dir, "serverCryptoIDChain.xml"), "r").read())
cryptoIDKey = parseXMLKey(open(os.path.join(dir, "serverCryptoIDKey.xml"), "r").read(), private=True)
connection = connect()
connection.handshakeClientCert(cryptoIDChain, cryptoIDKey)
assert(isinstance(connection.session.serverCertChain, CertChain))
assert(connection.session.serverCertChain.validate())
connection.close()
connection.sock.close()
print "Test 17 - mutual cryptoID faults"
for fault in Fault.clientCertFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeClientCert(cryptoIDChain, cryptoIDKey)
print " Good Fault %s" % (Fault.faultNames[fault])
except TLSFaultError, e:
print " BAD FAULT %s: %s" % (Fault.faultNames[fault], str(e))
badFault = True
connection.sock.close()
print "Test 18 - good SRP, prepare to resume..."
connection = connect()
connection.handshakeClientSRP("test", "password")
connection.close()
connection.sock.close()
session = connection.session
print "Test 19 - resumption"
connection = connect()
connection.handshakeClientSRP("test", "garbage", session=session)
#Don't close! -- see below
print "Test 20 - invalidated resumption"
connection.sock.close() #Close the socket without a close_notify!
connection = connect()
try:
connection.handshakeClientSRP("test", "garbage", session=session)
assert()
except TLSRemoteAlert, alert:
if alert.description != AlertDescription.bad_record_mac:
raise
connection.sock.close()
print "Test 21 - HTTPS test X.509"
address = address[0], address[1]+1
if hasattr(socket, "timeout"):
timeoutEx = socket.timeout
else:
timeoutEx = socket.error
while 1:
try:
time.sleep(2)
htmlBody = open(os.path.join(dir, "index.html")).read()
fingerprint = None
for y in range(2):
h = HTTPTLSConnection(\
address[0], address[1], x509Fingerprint=fingerprint)
for x in range(3):
h.request("GET", "/index.html")
r = h.getresponse()
assert(r.status == 200)
s = r.read()
assert(s == htmlBody)
fingerprint = h.tlsSession.serverCertChain.getFingerprint()
assert(fingerprint)
time.sleep(2)
break
except timeoutEx:
print "timeout, retrying..."
pass
if cryptoIDlibLoaded:
print "Test 21a - HTTPS test SRP+cryptoID"
address = address[0], address[1]+1
if hasattr(socket, "timeout"):
timeoutEx = socket.timeout
else:
timeoutEx = socket.error
while 1:
try:
time.sleep(2) #Time to generate key and cryptoID
htmlBody = open(os.path.join(dir, "index.html")).read()
fingerprint = None
protocol = None
for y in range(2):
h = HTTPTLSConnection(\
address[0], address[1],
username="test", password="password",
cryptoID=fingerprint, protocol=protocol)
for x in range(3):
h.request("GET", "/index.html")
r = h.getresponse()
assert(r.status == 200)
s = r.read()
assert(s == htmlBody)
fingerprint = h.tlsSession.serverCertChain.cryptoID
assert(fingerprint)
protocol = "urn:whatever"
time.sleep(2)
break
except timeoutEx:
print "timeout, retrying..."
pass
address = address[0], address[1]+1
implementations = []
if cryptlibpyLoaded:
implementations.append("cryptlib")
if m2cryptoLoaded:
implementations.append("openssl")
if pycryptoLoaded:
implementations.append("pycrypto")
implementations.append("python")
print "Test 22 - different ciphers"
for implementation in implementations:
for cipher in ["aes128", "aes256", "rc4"]:
print "Test 22:",
connection = connect()
settings = HandshakeSettings()
settings.cipherNames = [cipher]
settings.cipherImplementations = [implementation, "python"]
connection.handshakeClientSharedKey("shared", "key", settings=settings)
print ("%s %s" % (connection.getCipherName(), connection.getCipherImplementation()))
connection.write("hello")
h = connection.read(min=5, max=5)
assert(h == "hello")
connection.close()
connection.sock.close()
print "Test 23 - throughput test"
for implementation in implementations:
for cipher in ["aes128", "aes256", "3des", "rc4"]:
if cipher == "3des" and implementation not in ("openssl", "cryptlib", "pycrypto"):
continue
print "Test 23:",
connection = connect()
settings = HandshakeSettings()
settings.cipherNames = [cipher]
settings.cipherImplementations = [implementation, "python"]
connection.handshakeClientSharedKey("shared", "key", settings=settings)
print ("%s %s:" % (connection.getCipherName(), connection.getCipherImplementation())),
startTime = time.clock()
connection.write("hello"*10000)
h = connection.read(min=50000, max=50000)
stopTime = time.clock()
print "100K exchanged at rate of %d bytes/sec" % int(100000/(stopTime-startTime))
assert(h == "hello"*10000)
connection.close()
connection.sock.close()
print "Test 24 - Internet servers test"
try:
i = IMAP4_TLS("cyrus.andrew.cmu.edu")
i.login("anonymous", "anonymous@anonymous.net")
i.logout()
print "Test 24: IMAP4 good"
p = POP3_TLS("pop.gmail.com")
p.quit()
print "Test 24: POP3 good"
except socket.error, e:
print "Non-critical error: socket error trying to reach internet server: ", e
if not badFault:
print "Test succeeded"
else:
print "Test failed"
def serverTest(address, dir):
#Split address into hostname/port tuple
address = address.split(":")
if len(address)==1:
address.append("4443")
address = ( address[0], int(address[1]) )
#Connect to server
lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
lsock.bind(address)
lsock.listen(5)
def connect():
return TLSConnection(lsock.accept()[0])
print "Test 1 - good shared key"
sharedKeyDB = SharedKeyDB()
sharedKeyDB["shared"] = "key"
sharedKeyDB["shared2"] = "key2"
connection = connect()
connection.handshakeServer(sharedKeyDB=sharedKeyDB)
connection.close()
connection.sock.close()
print "Test 2 - shared key faults"
for fault in Fault.clientSharedKeyFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(sharedKeyDB=sharedKeyDB)
assert()
except:
pass
connection.sock.close()
print "Test 3 - good SRP"
#verifierDB = tlslite.VerifierDB(os.path.join(dir, "verifierDB"))
#verifierDB.open()
verifierDB = VerifierDB()
verifierDB.create()
entry = VerifierDB.makeVerifier("test", "password", 1536)
verifierDB["test"] = entry
connection = connect()
connection.handshakeServer(verifierDB=verifierDB)
connection.close()
connection.sock.close()
print "Test 4 - SRP faults"
for fault in Fault.clientSrpFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(verifierDB=verifierDB)
assert()
except:
pass
connection.sock.close()
print "Test 5 - good SRP: unknown_srp_username idiom"
connection = connect()
connection.handshakeServer(verifierDB=verifierDB)
connection.close()
connection.sock.close()
print "Test 6 - good SRP: with X.509 cert"
x509Cert = X509().parse(open(os.path.join(dir, "serverX509Cert.pem")).read())
x509Chain = X509CertChain([x509Cert])
s = open(os.path.join(dir, "serverX509Key.pem")).read()
x509Key = parsePEMKey(s, private=True)
connection = connect()
connection.handshakeServer(verifierDB=verifierDB, \
certChain=x509Chain, privateKey=x509Key)
connection.close()
connection.sock.close()
print "Test 7 - X.509 with SRP faults"
for fault in Fault.clientSrpFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(verifierDB=verifierDB, \
certChain=x509Chain, privateKey=x509Key)
assert()
except:
pass
connection.sock.close()
if cryptoIDlibLoaded:
print "Test 8 - good SRP: with cryptoID certs"
cryptoIDChain = CertChain().parse(open(os.path.join(dir, "serverCryptoIDChain.xml"), "r").read())
cryptoIDKey = parseXMLKey(open(os.path.join(dir, "serverCryptoIDKey.xml"), "r").read(), private=True)
connection = connect()
connection.handshakeServer(verifierDB=verifierDB, \
certChain=cryptoIDChain, privateKey=cryptoIDKey)
connection.close()
connection.sock.close()
print "Test 9 - cryptoID with SRP faults"
for fault in Fault.clientSrpFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(verifierDB=verifierDB, \
certChain=cryptoIDChain, privateKey=cryptoIDKey)
assert()
except:
pass
connection.sock.close()
print "Test 10 - good X.509"
connection = connect()
connection.handshakeServer(certChain=x509Chain, privateKey=x509Key)
connection.close()
connection.sock.close()
print "Test 10.a - good X.509, SSL v3"
connection = connect()
settings = HandshakeSettings()
settings.minVersion = (3,0)
settings.maxVersion = (3,0)
connection.handshakeServer(certChain=x509Chain, privateKey=x509Key, settings=settings)
connection.close()
connection.sock.close()
print "Test 11 - X.509 faults"
for fault in Fault.clientNoAuthFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(certChain=x509Chain, privateKey=x509Key)
assert()
except:
pass
connection.sock.close()
if cryptoIDlibLoaded:
print "Test 12 - good cryptoID"
connection = connect()
connection.handshakeServer(certChain=cryptoIDChain, privateKey=cryptoIDKey)
connection.close()
connection.sock.close()
print "Test 13 - cryptoID faults"
for fault in Fault.clientNoAuthFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(certChain=cryptoIDChain, privateKey=cryptoIDKey)
assert()
except:
pass
connection.sock.close()
print "Test 14 - good mutual X.509"
connection = connect()
connection.handshakeServer(certChain=x509Chain, privateKey=x509Key, reqCert=True)
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 14a - good mutual X.509, SSLv3"
connection = connect()
settings = HandshakeSettings()
settings.minVersion = (3,0)
settings.maxVersion = (3,0)
connection.handshakeServer(certChain=x509Chain, privateKey=x509Key, reqCert=True, settings=settings)
assert(isinstance(connection.session.serverCertChain, X509CertChain))
connection.close()
connection.sock.close()
print "Test 15 - mutual X.509 faults"
for fault in Fault.clientCertFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(certChain=x509Chain, privateKey=x509Key, reqCert=True)
assert()
except:
pass
connection.sock.close()
if cryptoIDlibLoaded:
print "Test 16 - good mutual cryptoID"
connection = connect()
connection.handshakeServer(certChain=cryptoIDChain, privateKey=cryptoIDKey, reqCert=True)
assert(isinstance(connection.session.serverCertChain, CertChain))
assert(connection.session.serverCertChain.validate())
connection.close()
connection.sock.close()
print "Test 17 - mutual cryptoID faults"
for fault in Fault.clientCertFaults + Fault.genericFaults:
connection = connect()
connection.fault = fault
try:
connection.handshakeServer(certChain=cryptoIDChain, privateKey=cryptoIDKey, reqCert=True)
assert()
except:
pass
connection.sock.close()
print "Test 18 - good SRP, prepare to resume"
sessionCache = SessionCache()
connection = connect()
connection.handshakeServer(verifierDB=verifierDB, sessionCache=sessionCache)
connection.close()
connection.sock.close()
print "Test 19 - resumption"
connection = connect()
connection.handshakeServer(verifierDB=verifierDB, sessionCache=sessionCache)
#Don't close! -- see next test
print "Test 20 - invalidated resumption"
try:
connection.read(min=1, max=1)
assert() #Client is going to close the socket without a close_notify
except TLSAbruptCloseError, e:
pass
connection = connect()
try:
connection.handshakeServer(verifierDB=verifierDB, sessionCache=sessionCache)
except TLSLocalAlert, alert:
if alert.description != AlertDescription.bad_record_mac:
raise
connection.sock.close()
print "Test 21 - HTTPS test X.509"
#Close the current listening socket
lsock.close()
#Create and run an HTTP Server using TLSSocketServerMixIn
class MyHTTPServer(TLSSocketServerMixIn,
BaseHTTPServer.HTTPServer):
def handshake(self, tlsConnection):
tlsConnection.handshakeServer(certChain=x509Chain, privateKey=x509Key)
return True
cd = os.getcwd()
os.chdir(dir)
address = address[0], address[1]+1
httpd = MyHTTPServer(address, SimpleHTTPServer.SimpleHTTPRequestHandler)
for x in range(6):
httpd.handle_request()
httpd.server_close()
cd = os.chdir(cd)
if cryptoIDlibLoaded:
print "Test 21a - HTTPS test SRP+cryptoID"
#Create and run an HTTP Server using TLSSocketServerMixIn
class MyHTTPServer(TLSSocketServerMixIn,
BaseHTTPServer.HTTPServer):
def handshake(self, tlsConnection):
tlsConnection.handshakeServer(certChain=cryptoIDChain, privateKey=cryptoIDKey,
verifierDB=verifierDB)
return True
cd = os.getcwd()
os.chdir(dir)
address = address[0], address[1]+1
httpd = MyHTTPServer(address, SimpleHTTPServer.SimpleHTTPRequestHandler)
for x in range(6):
httpd.handle_request()
httpd.server_close()
cd = os.chdir(cd)
#Re-connect the listening socket
lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
address = address[0], address[1]+1
lsock.bind(address)
lsock.listen(5)
def connect():
return TLSConnection(lsock.accept()[0])
implementations = []
if cryptlibpyLoaded:
implementations.append("cryptlib")
if m2cryptoLoaded:
implementations.append("openssl")
if pycryptoLoaded:
implementations.append("pycrypto")
implementations.append("python")
print "Test 22 - different ciphers"
for implementation in ["python"] * len(implementations):
for cipher in ["aes128", "aes256", "rc4"]:
print "Test 22:",
connection = connect()
settings = HandshakeSettings()
settings.cipherNames = [cipher]
settings.cipherImplementations = [implementation, "python"]
connection.handshakeServer(sharedKeyDB=sharedKeyDB, settings=settings)
print connection.getCipherName(), connection.getCipherImplementation()
h = connection.read(min=5, max=5)
assert(h == "hello")
connection.write(h)
connection.close()
connection.sock.close()
print "Test 23 - throughput test"
for implementation in implementations:
for cipher in ["aes128", "aes256", "3des", "rc4"]:
if cipher == "3des" and implementation not in ("openssl", "cryptlib", "pycrypto"):
continue
print "Test 23:",
connection = connect()
settings = HandshakeSettings()
settings.cipherNames = [cipher]
settings.cipherImplementations = [implementation, "python"]
connection.handshakeServer(sharedKeyDB=sharedKeyDB, settings=settings)
print connection.getCipherName(), connection.getCipherImplementation()
h = connection.read(min=50000, max=50000)
assert(h == "hello"*10000)
connection.write(h)
connection.close()
connection.sock.close()
print "Test succeeded"
if len(sys.argv) == 1 or (len(sys.argv)==2 and sys.argv[1].lower().endswith("help")):
print ""
print "Version: 0.3.8"
print ""
print "RNG: %s" % prngName
print ""
print "Modules:"
if cryptlibpyLoaded:
print " cryptlib_py : Loaded"
else:
print " cryptlib_py : Not Loaded"
if m2cryptoLoaded:
print " M2Crypto : Loaded"
else:
print " M2Crypto : Not Loaded"
if pycryptoLoaded:
print " pycrypto : Loaded"
else:
print " pycrypto : Not Loaded"
if gmpyLoaded:
print " GMPY : Loaded"
else:
print " GMPY : Not Loaded"
if cryptoIDlibLoaded:
print " cryptoIDlib : Loaded"
else:
print " cryptoIDlib : Not Loaded"
print ""
print "Commands:"
print ""
print " clientcert <server> [<chain> <key>]"
print " clientsharedkey <server> <user> <pass>"
print " clientsrp <server> <user> <pass>"
print " clienttest <server> <dir>"
print ""
print " serversrp <server> <verifierDB>"
print " servercert <server> <chain> <key> [req]"
print " serversrpcert <server> <verifierDB> <chain> <key>"
print " serversharedkey <server> <sharedkeyDB>"
print " servertest <server> <dir>"
sys.exit()
cmd = sys.argv[1].lower()
class Args:
def __init__(self, argv):
self.argv = argv
def get(self, index):
if len(self.argv)<=index:
raise SyntaxError("Not enough arguments")
return self.argv[index]
def getLast(self, index):
if len(self.argv)>index+1:
raise SyntaxError("Too many arguments")
return self.get(index)
args = Args(sys.argv)
def reformatDocString(s):
lines = s.splitlines()
newLines = []
for line in lines:
newLines.append(" " + line.strip())
return "\n".join(newLines)
try:
if cmd == "clienttest":
address = args.get(2)
dir = args.getLast(3)
clientTest(address, dir)
sys.exit()
elif cmd.startswith("client"):
address = args.get(2)
#Split address into hostname/port tuple
address = address.split(":")
if len(address)==1:
address.append("4443")
address = ( address[0], int(address[1]) )
def connect():
#Connect to server
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if hasattr(sock, "settimeout"):
sock.settimeout(5)
sock.connect(address)
#Instantiate TLSConnections
return TLSConnection(sock)
try:
if cmd == "clientsrp":
username = args.get(3)
password = args.getLast(4)
connection = connect()
start = time.clock()
connection.handshakeClientSRP(username, password)
elif cmd == "clientsharedkey":
username = args.get(3)
password = args.getLast(4)
connection = connect()
start = time.clock()
connection.handshakeClientSharedKey(username, password)
elif cmd == "clientcert":
certChain = None
privateKey = None
if len(sys.argv) > 3:
certFilename = args.get(3)
keyFilename = args.getLast(4)
s1 = open(certFilename, "rb").read()
s2 = open(keyFilename, "rb").read()
#Try to create cryptoID cert chain
if cryptoIDlibLoaded:
try:
certChain = CertChain().parse(s1)
privateKey = parsePrivateKey(s2)
except:
certChain = None
privateKey = None
#Try to create X.509 cert chain
if not certChain:
x509 = X509()
x509.parse(s1)
certChain = X509CertChain([x509])
privateKey = parsePrivateKey(s2)
connection = connect()
start = time.clock()
connection.handshakeClientCert(certChain, privateKey)
else:
raise SyntaxError("Unknown command")
except TLSLocalAlert, a:
if a.description == AlertDescription.bad_record_mac:
if cmd == "clientsharedkey":
print "Bad sharedkey password"
else:
raise
elif a.description == AlertDescription.user_canceled:
print str(a)
else:
raise
sys.exit()
except TLSRemoteAlert, a:
if a.description == AlertDescription.unknown_srp_username:
if cmd == "clientsrp":
print "Unknown username"
else:
raise
elif a.description == AlertDescription.bad_record_mac:
if cmd == "clientsrp":
print "Bad username or password"
else:
raise
elif a.description == AlertDescription.handshake_failure:
print "Unable to negotiate mutually acceptable parameters"
else:
raise
sys.exit()
stop = time.clock()
print "Handshake success"
print " Handshake time: %.4f seconds" % (stop - start)
print " Version: %s.%s" % connection.version
print " Cipher: %s %s" % (connection.getCipherName(), connection.getCipherImplementation())
if connection.session.srpUsername:
print " Client SRP username: %s" % connection.session.srpUsername
if connection.session.sharedKeyUsername:
print " Client shared key username: %s" % connection.session.sharedKeyUsername
if connection.session.clientCertChain:
print " Client fingerprint: %s" % connection.session.clientCertChain.getFingerprint()
if connection.session.serverCertChain:
print " Server fingerprint: %s" % connection.session.serverCertChain.getFingerprint()
connection.close()
connection.sock.close()
elif cmd.startswith("server"):
address = args.get(2)
#Split address into hostname/port tuple
address = address.split(":")
if len(address)==1:
address.append("4443")
address = ( address[0], int(address[1]) )
verifierDBFilename = None
sharedKeyDBFilename = None
certFilename = None
keyFilename = None
sharedKeyDB = None
reqCert = False
if cmd == "serversrp":
verifierDBFilename = args.getLast(3)
elif cmd == "servercert":
certFilename = args.get(3)
keyFilename = args.get(4)
if len(sys.argv)>=6:
req = args.getLast(5)
if req.lower() != "req":
raise SyntaxError()
reqCert = True
elif cmd == "serversrpcert":
verifierDBFilename = args.get(3)
certFilename = args.get(4)
keyFilename = args.getLast(5)
elif cmd == "serversharedkey":
sharedKeyDBFilename = args.getLast(3)
elif cmd == "servertest":
address = args.get(2)
dir = args.getLast(3)
serverTest(address, dir)
sys.exit()
verifierDB = None
if verifierDBFilename:
verifierDB = VerifierDB(verifierDBFilename)
verifierDB.open()
sharedKeyDB = None
if sharedKeyDBFilename:
sharedKeyDB = SharedKeyDB(sharedKeyDBFilename)
sharedKeyDB.open()
certChain = None
privateKey = None
if certFilename:
s1 = open(certFilename, "rb").read()
s2 = open(keyFilename, "rb").read()
#Try to create cryptoID cert chain
if cryptoIDlibLoaded:
try:
certChain = CertChain().parse(s1)
privateKey = parsePrivateKey(s2)
except:
certChain = None
privateKey = None
#Try to create X.509 cert chain
if not certChain:
x509 = X509()
x509.parse(s1)
certChain = X509CertChain([x509])
privateKey = parsePrivateKey(s2)
#Create handler function - performs handshake, then echos all bytes received
def handler(sock):
try:
connection = TLSConnection(sock)
settings = HandshakeSettings()
connection.handshakeServer(sharedKeyDB=sharedKeyDB, verifierDB=verifierDB, \
certChain=certChain, privateKey=privateKey, \
reqCert=reqCert, settings=settings)
print "Handshake success"
print " Version: %s.%s" % connection.version
print " Cipher: %s %s" % (connection.getCipherName(), connection.getCipherImplementation())
if connection.session.srpUsername:
print " Client SRP username: %s" % connection.session.srpUsername
if connection.session.sharedKeyUsername:
print " Client shared key username: %s" % connection.session.sharedKeyUsername
if connection.session.clientCertChain:
print " Client fingerprint: %s" % connection.session.clientCertChain.getFingerprint()
if connection.session.serverCertChain:
print " Server fingerprint: %s" % connection.session.serverCertChain.getFingerprint()
s = ""
while 1:
newS = connection.read()
if not newS:
break
s += newS
if s[-1]=='\n':
connection.write(s)
s = ""
except TLSLocalAlert, a:
if a.description == AlertDescription.unknown_srp_username:
print "Unknown SRP username"
elif a.description == AlertDescription.bad_record_mac:
if cmd == "serversrp" or cmd == "serversrpcert":
print "Bad SRP password for:", connection.allegedSrpUsername
else:
raise
elif a.description == AlertDescription.handshake_failure:
print "Unable to negotiate mutually acceptable parameters"
else:
raise
except TLSRemoteAlert, a:
if a.description == AlertDescription.bad_record_mac:
if cmd == "serversharedkey":
print "Bad sharedkey password for:", connection.allegedSharedKeyUsername
else:
raise
elif a.description == AlertDescription.user_canceled:
print "Handshake cancelled"
elif a.description == AlertDescription.handshake_failure:
print "Unable to negotiate mutually acceptable parameters"
elif a.description == AlertDescription.close_notify:
pass
else:
raise
#Run multi-threaded server
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(address)
sock.listen(5)
while 1:
(newsock, cliAddress) = sock.accept()
thread.start_new_thread(handler, (newsock,))
else:
print "Bad command: '%s'" % cmd
except TLSRemoteAlert, a:
print str(a)
raise