Add tests for TLS fallback on connection reset and close.

third_party/tlslite/tlslite/tlsconnection.py

 # Authors: 
 #   Trevor Perrin
 #   Google - added reqCAs parameter
 #   Google (adapted by Sam Rushing and Marcelo Fernandez) - NPN support
 #   Dimitris Moraitis - Anon ciphersuites
 #   Martin von Loewis - python 3 port
 #
 # See the LICENSE file for legal information regarding use of this file.
 
 """
@@ skipping 1047 matching lines @@
     # Server Handshake Functions
     #*********************************************************
 
 
     def handshakeServer(self, verifierDB=None,
                         certChain=None, privateKey=None, reqCert=False,
                         sessionCache=None, settings=None, checker=None,
                         reqCAs = None, reqCertTypes = None,
                         tacks=None, activationFlags=0,
                         nextProtos=None, anon=False,
-                        tlsIntolerant=None, signedCertTimestamps=None,
+                        signedCertTimestamps=None,
                         fallbackSCSV=False, ocspResponse=None):
         """Perform a handshake in the role of server.
 
         This function performs an SSL or TLS handshake.  Depending on
         the arguments and the behavior of the client, this function can
         perform an SRP, or certificate-based handshake.  It
         can also perform a combined SRP and server-certificate
         handshake.
 
         Like any handshake function, this can be called on a closed
@@ skipping 53 matching lines @@
 
         @type reqCertTypes: list of int
         @param reqCertTypes: A list of certificate_type values to be sent
         along with a certificate request. This does not affect verification.
 
         @type nextProtos: list of strings.
         @param nextProtos: A list of upper layer protocols to expose to the
         clients through the Next-Protocol Negotiation Extension, 
         if they support it.
 
-        @type tlsIntolerant: (int, int) or None
-        @param tlsIntolerant: If tlsIntolerant is not None, the server will
-        simulate TLS version intolerance by returning a fatal handshake_failure
-        alert to all TLS versions tlsIntolerant or higher.
-
         @type signedCertTimestamps: str
         @param signedCertTimestamps: A SignedCertificateTimestampList (as a
         binary 8-bit string) that will be sent as a TLS extension whenever
         the client announces support for the extension.
 
         @type fallbackSCSV: bool
         @param fallbackSCSV: if true, the server will implement
         TLS_FALLBACK_SCSV and thus reject connections using less than the
         server's maximum TLS version that include this cipher suite.
 
@@ skipping 11 matching lines @@
         @raise tlslite.errors.TLSAbruptCloseError: If the socket is closed
         without a preceding alert.
         @raise tlslite.errors.TLSAlert: If a TLS alert is signalled.
         @raise tlslite.errors.TLSAuthenticationError: If the checker
         doesn't like the other party's authentication credentials.
         """
         for result in self.handshakeServerAsync(verifierDB,
                 certChain, privateKey, reqCert, sessionCache, settings,
                 checker, reqCAs, reqCertTypes,
                 tacks=tacks, activationFlags=activationFlags, 
-                nextProtos=nextProtos, anon=anon, tlsIntolerant=tlsIntolerant,
+                nextProtos=nextProtos, anon=anon,
                 signedCertTimestamps=signedCertTimestamps,
                 fallbackSCSV=fallbackSCSV, ocspResponse=ocspResponse):
             pass
 
 
     def handshakeServerAsync(self, verifierDB=None,
                              certChain=None, privateKey=None, reqCert=False,
                              sessionCache=None, settings=None, checker=None,
                              reqCAs=None, reqCertTypes=None,
                              tacks=None, activationFlags=0,
                              nextProtos=None, anon=False,
-                             tlsIntolerant=None,
                              signedCertTimestamps=None,
                              fallbackSCSV=False,
                              ocspResponse=None
                              ):
         """Start a server handshake operation on the TLS connection.
 
         This function returns a generator which behaves similarly to
         handshakeServer().  Successive invocations of the generator
         will return 0 if it is waiting to read from the socket, 1 if it is
         waiting to write to the socket, or it will raise StopIteration
         if the handshake operation is complete.
 
         @rtype: iterable
         @return: A generator; see above for details.
         """
         handshaker = self._handshakeServerAsyncHelper(\
             verifierDB=verifierDB, certChain=certChain,
             privateKey=privateKey, reqCert=reqCert,
             sessionCache=sessionCache, settings=settings, 
             reqCAs=reqCAs, reqCertTypes=reqCertTypes,
             tacks=tacks, activationFlags=activationFlags, 
             nextProtos=nextProtos, anon=anon,
-            tlsIntolerant=tlsIntolerant,
             signedCertTimestamps=signedCertTimestamps,
             fallbackSCSV=fallbackSCSV,
             ocspResponse=ocspResponse)
         for result in self._handshakeWrapperAsync(handshaker, checker):
             yield result
 
 
     def _handshakeServerAsyncHelper(self, verifierDB,
                              certChain, privateKey, reqCert, sessionCache,
                              settings, reqCAs, reqCertTypes,
                              tacks, activationFlags, 
                              nextProtos, anon,
-                             tlsIntolerant, signedCertTimestamps, fallbackSCSV,
+                             signedCertTimestamps, fallbackSCSV,
                              ocspResponse):
 
         self._handshakeStart(client=False)
 
         if (not verifierDB) and (not certChain) and not anon:
             raise ValueError("Caller passed no authentication credentials")
         if certChain and not privateKey:
             raise ValueError("Caller passed a certChain but no privateKey")
         if privateKey and not certChain:
             raise ValueError("Caller passed a privateKey but no certChain")
@@ skipping 17 matching lines @@
         if not settings:
             settings = HandshakeSettings()
         settings = settings._filter()
         
         # OK Start exchanging messages
         # ******************************
         
         # Handle ClientHello and resumption
         for result in self._serverGetClientHello(settings, certChain,\
                                             verifierDB, sessionCache,
-                                            anon, tlsIntolerant, fallbackSCSV):
+                                            anon, fallbackSCSV):
             if result in (0,1): yield result
             elif result == None:
                 self._handshakeDone(resumed=True)                
                 return # Handshake was resumed, we're done 
             else: break
         (clientHello, cipherSuite) = result
         
         #If not a resumption...
 
         # Create the ServerHello message
@@ skipping 94 matching lines @@
             tackExt, serverHello.tackExt!=None, serverName)
             
         #Add the session object to the session cache
         if sessionCache and sessionID:
             sessionCache[sessionID] = self.session
 
         self._handshakeDone(resumed=False)
 
 
     def _serverGetClientHello(self, settings, certChain, verifierDB,
-                                sessionCache, anon, tlsIntolerant, fallbackSCSV):
+                                sessionCache, anon, fallbackSCSV):
         #Initialize acceptable cipher suites
         cipherSuites = []
         if verifierDB:
             if certChain:
                 cipherSuites += \
                     CipherSuite.getSrpCertSuites(settings)
             cipherSuites += CipherSuite.getSrpSuites(settings)
         elif certChain:
             cipherSuites += CipherSuite.getCertSuites(settings)
             cipherSuites += CipherSuite.getDheCertSuites(settings)
@@ skipping 16 matching lines @@
 
         #If client's version is too low, reject it
         if clientHello.client_version < settings.minVersion:
             self.version = settings.minVersion
             for result in self._sendError(\
                   AlertDescription.protocol_version,
                   "Too old version: %s" % str(clientHello.client_version)):
                 yield result
 
         #If simulating TLS intolerance, reject certain TLS versions.
-        elif (tlsIntolerant is not None and
-            clientHello.client_version >= tlsIntolerant):
-            for result in self._sendError(\
+        elif (settings.tlsIntolerant is not None and
+              clientHello.client_version >= settings.tlsIntolerant):
+            if settings.tlsIntoleranceType == "alert":
+                for result in self._sendError(\
                     AlertDescription.handshake_failure):
-                yield result
+                    yield result

[wtc, 2014/06/24 21:33:29]

Should we also call
    raise TLSUnsupportedError("Simulating version intolerance")
in this case?

[davidben, 2014/06/25 21:19:57]

self._sendError actually raises TLSLocalAlert for you.

+            elif settings.tlsIntoleranceType == "close":
+                self._abruptClose()
+                raise TLSUnsupportedError("Simulating version intolerance")
+            elif settings.tlsIntoleranceType == "reset":
+                self._abruptClose(reset=True)
+                raise TLSUnsupportedError("Simulating version intolerance")
+            else:
+                raise ValueError("Unknown intolerance type: '%s'" %
+                                 settings.tlsIntoleranceType)
 
         #If client's version is too high, propose my highest version
         elif clientHello.client_version > settings.maxVersion:
             self.version = settings.maxVersion
 
         #Detect if the client performed an inappropriate fallback.
         elif fallbackSCSV and clientHello.client_version < settings.maxVersion:
             if CipherSuite.TLS_FALLBACK_SCSV in clientHello.cipher_suites:
                 for result in self._sendError(\
                         AlertDescription.inappropriate_fallback):
@@ skipping 494 matching lines @@
             except TLSAlert as alert:
                 if not self.fault:
                     raise
                 if alert.description not in Fault.faultAlerts[self.fault]:
                     raise TLSFaultError(str(alert))
                 else:
                     pass
             except:
                 self._shutdown(False)
                 raise