add support for requiring certificate presence

Author: tomato42

tests/tlstest.py

@@ -723,6 +723,27 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - mutual X.509, PHA, no client cert, TLSv1.3".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 4)
+    settings.maxVersion = (3, 4)
+    connection.handshakeClientCert(X509CertChain(), x509Key, settings=settings)
+    synchro.recv(1)
+    b = connection.read(0, 0)
+    assert b == b''
+    try:
+        connection.read(0, 0)
+        assert False
+    except TLSRemoteAlert as e:
+        assert e.description == AlertDescription.certificate_required
+        assert "certificate_required" in str(e), str(e)
+
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - good mutual X.509, TLSv1.1".format(test_no))
     synchro.recv(1)
     connection = connect()
@@ -1970,6 +1991,32 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - mutual X.509, PHA, no client cert, TLSv1.3".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 4)
+    settings.maxVersion = (3, 4)
+    connection.handshakeServer(certChain=x509Chain, privateKey=x509Key,
+                               settings=settings)
+    connection.client_cert_required = True
+    assert connection.session.clientCertChain is None
+    for result in connection.request_post_handshake_auth(settings):
+        assert result in (0, 1)
+    synchro.send(b'R')
+    try:
+        testConnServer(connection)
+        assert False
+    except TLSLocalAlert as e:
+        assert "Client did not provide a certificate in post-handshake" in \
+            str(e)
+        assert e.description == AlertDescription.certificate_required
+
+    assert connection.session.clientCertChain is None
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - good mutual X.509, TLSv1.1".format(test_no))
     synchro.send(b'R')
     connection = connect()

tlslite/constants.py

@@ -488,6 +488,7 @@ class AlertDescription(TLSEnum):
     bad_certificate_status_response = 113  # RFC 6066
     bad_certificate_hash_value = 114  # RFC 6066
     unknown_psk_identity = 115
+    certificate_required = 116  # RFC 8446
     no_application_protocol = 120  # RFC 7301
 
 

tlslite/errors.py

@@ -59,34 +59,6 @@ class TLSAlert(TLSError):
 
     pass
 
-    _descriptionStr = {\
-        AlertDescription.close_notify: "close_notify",\
-        AlertDescription.unexpected_message: "unexpected_message",\
-        AlertDescription.bad_record_mac: "bad_record_mac",\
-        AlertDescription.decryption_failed: "decryption_failed",\
-        AlertDescription.record_overflow: "record_overflow",\
-        AlertDescription.decompression_failure: "decompression_failure",\
-        AlertDescription.handshake_failure: "handshake_failure",\
-        AlertDescription.no_certificate: "no certificate",\
-        AlertDescription.bad_certificate: "bad_certificate",\
-        AlertDescription.unsupported_certificate: "unsupported_certificate",\
-        AlertDescription.certificate_revoked: "certificate_revoked",\
-        AlertDescription.certificate_expired: "certificate_expired",\
-        AlertDescription.certificate_unknown: "certificate_unknown",\
-        AlertDescription.illegal_parameter: "illegal_parameter",\
-        AlertDescription.unknown_ca: "unknown_ca",\
-        AlertDescription.access_denied: "access_denied",\
-        AlertDescription.decode_error: "decode_error",\
-        AlertDescription.decrypt_error: "decrypt_error",\
-        AlertDescription.export_restriction: "export_restriction",\
-        AlertDescription.protocol_version: "protocol_version",\
-        AlertDescription.insufficient_security: "insufficient_security",\
-        AlertDescription.internal_error: "internal_error",\
-        AlertDescription.inappropriate_fallback: "inappropriate_fallback",\
-        AlertDescription.user_canceled: "user_canceled",\
-        AlertDescription.no_renegotiation: "no_renegotiation",\
-        AlertDescription.unknown_psk_identity: "unknown_psk_identity"}
-
 
 class TLSLocalAlert(TLSAlert):
     """A TLS alert has been signalled by the local implementation.
@@ -109,9 +81,7 @@ def __init__(self, alert, message=None):
         self.message = message
 
     def __str__(self):
-        alertStr = TLSAlert._descriptionStr.get(self.description)
-        if alertStr == None:
-            alertStr = str(self.description)
+        alertStr = AlertDescription.toStr(self.description)
         if self.message:
             return alertStr + ": " + self.message
         else:
@@ -136,9 +106,7 @@ def __init__(self, alert):
         self.level = alert.level
 
     def __str__(self):
-        alertStr = TLSAlert._descriptionStr.get(self.description)
-        if alertStr == None:
-            alertStr = str(self.description)
+        alertStr = AlertDescription.toStr(self.description)
         return alertStr
 
 

tlslite/tlsconnection.py

@@ -4212,7 +4212,7 @@ def _sigHashesToList(settings, privateKey=None, certList=None,
         """Convert list of valid signature hashes to array of tuples"""
         certType = None
         publicKey = None
-        if certList:
+        if certList and certList.x509List:
             certType = certList.x509List[0].certAlg
             publicKey = certList.x509List[0].publicKey
 

tlslite/tlsrecordlayer.py

@@ -111,6 +111,11 @@ class TLSRecordLayer(object):
 
     :vartype tickets: list of bytearray
     :ivar tickets: list of session tickets received from server, oldest first.
+
+    :vartype client_cert_required: bool
+    :ivar client_cert_required: Set to True to make the post-handshake
+        authentication fail when client doesn't provide a certificate in
+        response
     """
 
     def __init__(self, sock):
@@ -193,6 +198,10 @@ def __init__(self, sock):
         # messages (which are the values)
         self._cert_requests = {}
 
+        # boolean to control if PHA needs to be aborted when the client
+        # doesn't provide a certificate
+        self.client_cert_required = False
+
     @property
     def _send_record_limit(self):
         """Maximum size of payload that can be sent."""
@@ -663,7 +672,7 @@ def _handle_pha(self, cert_request):
         msgs.append(Certificate(CertificateType.x509, self.version)
                     .create(cert, cert_request.certificate_request_context))
         handshake_context.update(msgs[0].write())
-        if p_key:
+        if cert.x509List and p_key:
             # sign the CertificateVerify only when we have a private key to do
             # that
             valid_sig_algs = cert_request.supported_signature_algs
@@ -809,6 +818,12 @@ def _handle_srv_pha(self, cert):
                         "Signature verification failed"):
                     yield result
             handshake_context.update(cert_verify.write())
+        elif self.client_cert_required:
+            for result in self._sendError(
+                    AlertDescription.certificate_required,
+                    "Client did not provide a certificate in post-handshake "
+                    "authentication"):
+                yield result
 
         finished_key = HKDF_expand_label(self.session.cl_app_secret,
                                          b'finished', b'',