-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