tests for EC point format

tests/tlstest.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Authors: 
+# Authors:
 #   Trevor Perrin
 #   Kees Bos - Added tests for XML-RPC
 #   Dimitris Moraitis - Anon ciphersuites
@@ -48,26 +48,26 @@
 
 try:
     from tack.structures.Tack import Tack
-    
+
 except ImportError:
     pass
 
 def printUsage(s=None):
     if m2cryptoLoaded:
         crypto = "M2Crypto/OpenSSL"
     else:
-        crypto = "Python crypto"        
+        crypto = "Python crypto"
     if s:
         print("ERROR: %s" % s)
-    print("""\ntls.py version %s (using %s)  
+    print("""\ntls.py version %s (using %s)
 
 Commands:
   server HOST:PORT DIRECTORY
 
   client HOST:PORT DIRECTORY
 """ % (__version__, crypto))
     sys.exit(-1)
-    
+
 
 def testConnClient(conn):
     b1 = os.urandom(1)
@@ -92,9 +92,9 @@ def testConnClient(conn):
     assert r1000 == b1000
 
 def clientTestCmd(argv):
-    
+
     address = argv[0]
-    dir = argv[1]    
+    dir = argv[1]
 
     #Split address into hostname/port tuple
     address = address.split(":")
@@ -235,7 +235,7 @@ def connect():
     settings.minVersion = (3,0)
     settings.maxVersion = (3,0)
     connection.handshakeClientCert(settings=settings)
-    testConnClient(connection)    
+    testConnClient(connection)
     assert(isinstance(connection.session.serverCertChain, X509CertChain))
     connection.close()
 
@@ -307,14 +307,51 @@ def connect():
     settings.minVersion = (3, 3)
     settings.maxVersion = (3, 3)
     settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
-    settings.ec_point_formats = [ECPointFormat.ansiX962_compressed_prime, ECPointFormat.uncompressed]
     connection.handshakeClientCert(settings=settings)
     testConnClient(connection)
     assert connection.session.ec_point_format == ECPointFormat.ansiX962_compressed_prime
     connection.close()
 
     test_no += 1
 
+    print("Test {0} - client uncompressed - error, TLSv1.2".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    settings.ec_point_formats = [ECPointFormat.uncompressed]
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    try:
+        connection.handshakeClientCert(settings=settings)
+        assert False
+    except TLSIllegalParameterException  as e:
+        assert "No common EC point format" in str(e)
+    except TLSAbruptCloseError as e:
+        pass
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - client comppressed char2 - error, TLSv1.2".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    settings.ec_point_formats = [ECPointFormat.ansiX962_compressed_char2]
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    try:
+        connection.handshakeClientCert(settings=settings)
+        assert False
+    except ValueError  as e:
+        assert "Unknown EC point format provided: [2]" in str(e)
+    except TLSAbruptCloseError as e:
+        pass
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - mismatched ECDSA curve, TLSv1.2".format(test_no))
     synchro.recv(1)
     connection = connect()
@@ -2191,7 +2228,7 @@ def connect():
 
     test_no += 1
 
-    print("Test {0} server uncompressed ec format - uncompressed, TLSv1.2".format(test_no))
+    print("Test {0} - server uncompressed ec format - uncompressed, TLSv1.2".format(test_no))
     synchro.send(b'R')
     connection = connect()
     settings = HandshakeSettings()
@@ -2207,23 +2244,60 @@ def connect():
 
     test_no += 1
 
-    print("Test {0} server compressed ec format - compressed, TLSv1.2".format(test_no))
+    print("Test {0} - server compressed ec format - compressed, TLSv1.2".format(test_no))
     synchro.send(b'R')
     connection = connect()
     settings = HandshakeSettings()
     settings.minVersion = (3, 1)
     settings.maxVersion = (3, 3)
     settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
-    settings.ec_point_formats = [ECPointFormat.ansiX962_compressed_prime, ECPointFormat.uncompressed]
     connection.handshakeServer(certChain=x509ecdsaChain,
                             privateKey=x509ecdsaKey, settings=settings)
     testConnServer(connection)
-    print(connection.session.ec_point_format)
     assert connection.session.ec_point_format == ECPointFormat.ansiX962_compressed_prime
     connection.close()
 
     test_no +=1
 
+    print("Test {0} - server compressed ec format - error, TLSv1.2".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 1)
+    settings.maxVersion = (3, 3)
+    settings.ec_point_formats = [ECPointFormat.ansiX962_compressed_prime]
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    try:
+        connection.handshakeServer(certChain=x509ecdsaChain,
+                                privateKey=x509ecdsaKey, settings=settings)
+        assert False
+    except TLSIllegalParameterException as e:
+        assert "No common EC point format" in str(e)
+    except TLSAbruptCloseError as e:
+        pass
+    connection.close()
+
+    test_no +=1
+
+    print("Test {0} - client compressed char2 - error, TLSv1.2".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 1)
+    settings.maxVersion = (3, 3)
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    try:
+        connection.handshakeServer(certChain=x509ecdsaChain,
+                                privateKey=x509ecdsaKey, settings=settings)
+        assert False
+    except ValueError as e:
+        assert "Unknown EC point format provided: [2]" in str(e)
+    except TLSAbruptCloseError as e:
+        pass
+    connection.close()
+
+    test_no +=1
+
     print("Test {0} - mismatched ECDSA curve, TLSv1.2".format(test_no))
     synchro.send(b'R')
     connection = connect()
@@ -2508,7 +2582,7 @@ def connect():
             connection.handshakeServer(certChain=x509Chain, privateKey=x509Key,
                 tacks=[tackUnrelated], settings=settings)
             assert False
-        except TLSRemoteAlert as alert:
+        except TLSLocalAlert as alert:
             if alert.description != AlertDescription.illegal_parameter:
                 raise
     else:

tlslite/handshakesettings.py

@@ -61,8 +61,7 @@
 TICKET_CIPHERS = ["chacha20-poly1305", "aes256gcm", "aes128gcm", "aes128ccm",
                   "aes128ccm_8", "aes256ccm", "aes256ccm_8"]
 PSK_MODES = ["psk_dhe_ke", "psk_ke"]
-EC_POINT_FORMATS = [ECPointFormat.ansiX962_compressed_char2,
-                    ECPointFormat.ansiX962_compressed_prime,
+EC_POINT_FORMATS = [ECPointFormat.ansiX962_compressed_prime,
                     ECPointFormat.uncompressed]
 
 

tlslite/keyexchange.py

@@ -709,10 +709,11 @@ def makeServerKeyExchange(self, sigHash=None):
         ext_c = self.clientHello.getExtension(ExtensionType.ec_point_formats)
         ext_s = self.serverHello.getExtension(ExtensionType.ec_point_formats)
         if ext_c and ext_s:
-            for ext in ext_c.formats:
-                if ext in ext_s.formats:
-                    ext_negotiated = ext
-                    break
+            try:
+                ext_negotiated = next((i for i in ext_c.formats \
+                                       if i in ext_s.formats))
+            except StopIteration:
+                raise TLSIllegalParameterException("No common EC point format")
 
         ecdhYs = kex.calc_public_value(self.ecdhXs, ext_negotiated)
 
@@ -739,6 +740,8 @@ def processClientKeyExchange(self, clientKeyExchange):
             ext_supported = [
                 ext for ext in ext_c.formats if ext in ext_s.formats
                 ]
+            if not ext_supported:
+                raise TLSIllegalParameterException("No common EC point format")
         return kex.calc_shared_key(self.ecdhXs, ecdhYc, ext_supported)
 
     def processServerKeyExchange(self, srvPublicKey, serverKeyExchange):
@@ -762,8 +765,11 @@ def processServerKeyExchange(self, srvPublicKey, serverKeyExchange):
         ext_c = self.clientHello.getExtension(ExtensionType.ec_point_formats)
         ext_s = self.serverHello.getExtension(ExtensionType.ec_point_formats)
         if ext_c and ext_s:
-            ext_supported = [i for i in ext_c.formats if i in ext_s.formats]
-            ext_negotiated = ext_supported[0]
+            try:
+                ext_supported = [i for i in ext_c.formats if i in ext_s.formats]
+                ext_negotiated = ext_supported[0]
+            except IndexError:
+                raise TLSIllegalParameterException("No common EC point format")
 
         self.ecdhYc = kex.calc_public_value(ecdhXc, ext_negotiated)
         return kex.calc_shared_key(ecdhXc, ecdh_Ys, ext_supported)

tlslite/tlsconnection.py

@@ -659,9 +659,16 @@ def _handshakeClientAsyncHelper(self, srpParams, certParams, anonParams,
         ext_s = serverHello.getExtension(ExtensionType.ec_point_formats)
         ext_ec_point = ECPointFormat.uncompressed
         if ext_c and ext_s:
-            ext_ec_point = next((i for i in ext_c.formats \
-                                 if i in ext_s.formats), \
-                                    ECPointFormat.uncompressed)
+            try:
+                ext_ec_point = next((i for i in ext_c.formats \
+                                     if i in ext_s.formats))
+
+            except StopIteration as alert:
+                for result in self._sendError(
+                        AlertDescription.illegal_parameter,
+                        str(alert)):
+                    yield result
+                pass
 
         # Create the session object which is used for resumptions
         self.session = Session()
@@ -2424,9 +2431,15 @@ def _handshakeServerAsyncHelper(self, verifierDB,
         ext_s = serverHello.getExtension(ExtensionType.ec_point_formats)
         ext_ec_point = ECPointFormat.uncompressed
         if ext_c and ext_s:
-            ext_ec_point = next((i for i in ext_c.formats \
-                                 if i in ext_s.formats),\
-                                      ECPointFormat.uncompressed)
+            try:
+                ext_ec_point = next((i for i in ext_c.formats \
+                                    if i in ext_s.formats))
+            except StopIteration as alert:
+                for result in self._sendError(
+                        AlertDescription.illegal_parameter,
+                        str(alert)):
+                    yield result
+                pass
 
         # We'll update the session master secret once it is calculated
         # in _serverFinished