Point extension

.gitignore

@@ -9,4 +9,4 @@ coverage.xml
 pylint_report.txt
 build/
 docs/_build/
-htmlcov/
+htmlcov/

scripts/tls.py

@@ -406,6 +406,7 @@ def printGoodConnection(connection, seconds):
     if connection.server_cert_compression_algo:
         print("  Server compression algorithm used: {0}".format(
             connection.server_cert_compression_algo))
+    print("  Session used ec point format extension: {0}".format(connection.session.ec_point_format))
 
 def printExporter(connection, expLabel, expLength):
     if expLabel is None:

tests/tlstest.py

@@ -44,7 +44,7 @@
     from xmlrpc import client as xmlrpclib
 import ssl
 from tlslite import *
-from tlslite.constants import KeyUpdateMessageType, SignatureScheme
+from tlslite.constants import KeyUpdateMessageType, ECPointFormat, SignatureScheme
 
 try:
     from tack.structures.Tack import Tack
@@ -303,6 +303,76 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - client compressed/uncompressed - uncompressed, TLSv1.2".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    settings.keyShares = ["secp256r1"]
+    connection.handshakeClientCert(settings=settings)
+    testConnClient(connection)
+    assert connection.session.ec_point_format == ECPointFormat.uncompressed
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - client compressed - compressed, TLSv1.2".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    settings.keyShares = ["secp256r1"]
+    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 missing 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.ansiX962_compressed_prime]
+    settings.eccCurves = ["secp256r1", "secp384r1", "secp521r1", "x25519", "x448"]
+    settings.keyShares = ["secp256r1"]
+    try:
+        connection.handshakeClientCert(settings=settings)
+        assert False
+    except ValueError  as e:
+        assert "Uncompressed EC point format is not provided" 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"]
+    settings.keyShares = ["secp256r1"]
+    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()
@@ -2220,6 +2290,79 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - server uncompressed ec format - uncompressed, 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.keyShares = ["secp256r1"]
+    settings.ec_point_formats = [ECPointFormat.uncompressed]
+    connection.handshakeServer(certChain=x509ecdsaChain,
+                            privateKey=x509ecdsaKey, settings=settings)
+    testConnServer(connection)
+    assert connection.session.ec_point_format == ECPointFormat.uncompressed
+    connection.close()
+
+    test_no += 1
+
+    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.keyShares = ["secp256r1"]
+    connection.handshakeServer(certChain=x509ecdsaChain,
+                            privateKey=x509ecdsaKey, settings=settings)
+    testConnServer(connection)
+    assert connection.session.ec_point_format == ECPointFormat.ansiX962_compressed_prime
+    connection.close()
+
+    test_no +=1
+
+    print("Test {0} - server missing uncompressed in client - 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"]
+    settings.keyShares = ["secp256r1"]
+    try:
+        connection.handshakeServer(certChain=x509ecdsaChain,
+                                privateKey=x509ecdsaKey, settings=settings)
+        assert False
+    except ValueError as e:
+        assert "Uncompressed EC point format is not provided" 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"]
+    settings.keyShares = ["secp256r1"]
+    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()
@@ -3509,7 +3652,7 @@ def heartbeat_response_check(message):
     assert synchro.recv(1) == b'R'
     connection.close()
 
-    test_no += 1
+    test_no +=1
 
     print("Tests {0}-{1} - XMLRPXC server".format(test_no, test_no + 2))
 
@@ -3542,6 +3685,7 @@ def add(self, x, y): return x + y
 
     synchro.close()
     synchroSocket.close()
+
     test_no += 2
 
     print("Test succeeded")

tlslite/handshakesettings.py

@@ -7,7 +7,7 @@
 
 """Class for setting handshake parameters."""
 
-from .constants import CertificateType
+from .constants import CertificateType, ECPointFormat
 from .utils import cryptomath
 from .utils import cipherfactory
 from .utils.compat import ecdsaAllCurves, int_types, ML_KEM_AVAILABLE
@@ -74,6 +74,8 @@
 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_prime,
+                    ECPointFormat.uncompressed]
 
 ALL_COMPRESSION_ALGOS_SEND = ["zlib"]
 if compression_algo_impls["brotli_compress"]:
@@ -395,6 +397,10 @@ class HandshakeSettings(object):
         option is for when a certificate was received/decompressed by this
         peer.
 
+
+    :vartype ec_point_formats: list
+    :ivar ec_point_formats: Enabled point format extension for
+     elliptic curves.
     """
 
     def _init_key_settings(self):
@@ -442,6 +448,7 @@ def _init_misc_extensions(self):
         # resumed connections (as tickets are single-use in TLS 1.3
         self.ticket_count = 2
         self.record_size_limit = 2**14 + 1  # TLS 1.3 includes content type
+        self.ec_point_formats = list(EC_POINT_FORMATS)
 
         # Certificate compression
         self.certificate_compression_send = list(ALL_COMPRESSION_ALGOS_SEND)
@@ -652,6 +659,14 @@ def _sanityCheckExtensions(other):
                 not 64 <= other.record_size_limit <= 2**14 + 1:
             raise ValueError("record_size_limit cannot exceed 2**14+1 bytes")
 
+        bad_ec_ext = [i for i in other.ec_point_formats if
+                      i not in EC_POINT_FORMATS]
+        if bad_ec_ext:
+            raise ValueError("Unknown EC point format provided: "
+                             "{0}".format(bad_ec_ext))
+        if ECPointFormat.uncompressed not in other.ec_point_formats:
+            raise ValueError("Uncompressed EC point format is not provided")
+
         HandshakeSettings._sanityCheckEMSExtension(other)
 
         if other.certificate_compression_send:
@@ -746,6 +761,7 @@ def _copy_extension_settings(self, other):
         other.sendFallbackSCSV = self.sendFallbackSCSV
         other.useEncryptThenMAC = self.useEncryptThenMAC
         other.usePaddingExtension = self.usePaddingExtension
+        other.ec_point_formats = self.ec_point_formats
         # session tickets
         other.padding_cb = self.padding_cb
         other.ticketKeys = self.ticketKeys