feat: Implement RFC 9345 Delegated Credentials

Adds support for Delegated Credentials (DC) on both server and client.

Includes DC class, parsing, extension handling, settings integration,
and validation.

Author: gstarovo

scripts/tls.py

@@ -40,6 +40,7 @@
 from tlslite.utils.cryptomath import getRandomBytes
 from tlslite.constants import KeyUpdateMessageType
 from tlslite.utils.compression import compression_algo_impls
+from tlslite.utils.pem import dePem
 
 try:
     from tack.structures.Tack import Tack
@@ -109,7 +110,7 @@ def printUsage(s=None):
     [-c CERT] [-k KEY] [-t TACK] [-v VERIFIERDB] [-d DIR] [-l LABEL] [-L LENGTH]
     [--reqcert] [--param DHFILE] [--psk PSK] [--psk-ident IDENTITY]
     [--psk-sha384] [--ssl3] [--max-ver VER] [--tickets COUNT] [--cipherlist]
-    [--request-pha] [--require-pha] [--echo] [--groups GROUPS]
+    [--request-pha] [--require-pha] [--echo] [--groups GROUPS] [--dc-key KEY]
     HOST:PORT
 
   client
@@ -138,6 +139,8 @@ def printUsage(s=None):
                   post-handshake authentication
   --echo - function as an echo server
   --groups - specify what key exchange groups should be supported
+  --dc-key KEY - the private key of the delegated credential
+  --dc-pub KEY - the public key of the delegated credential
   GROUPS - comma-separated list of enabled key exchange groups
   CERT, KEY - the file with key and certificates that will be used by client or
         server. The server can accept multiple pairs of `-c` and `-k` options
@@ -200,6 +203,8 @@ def handleArgs(argv, argString, flagsList=[]):
     require_pha = False
     echo = False
     groups = None
+    dc_key = None
+    dc_pub = None
 
     for opt, arg in opts:
         if opt == "-k":
@@ -231,6 +236,20 @@ def handleArgs(argv, argString, flagsList=[]):
             else:
                 v_host_cert = X509CertChain()
                 v_host_cert.parsePemList(s)
+        elif opt == "--dc-key":
+            s = open(arg, "rb").read()
+            if sys.version_info[0] >= 3:
+                s = str(s, 'utf-8')
+            if not cert_chain:
+                raise ValueError("Certificate is missing (must be listed "
+                                 "before the delegated credentials)")
+            dc_key = parsePEMKey(s, private=True,
+                                      implementations=["python"])
+        elif opt == "--dc-pub":
+            s = open(arg, "rb").read()
+            if sys.version_info[0] >= 3:
+                s = str(s, 'utf-8')
+            dc_pub = dePem(s, "PUBLIC KEY")
         elif opt == "-u":
             username = arg
         elif opt == "-p":
@@ -351,6 +370,10 @@ def handleArgs(argv, argString, flagsList=[]):
         retList.append(echo)
     if "groups=" in flagsList:
         retList.append(groups)
+    if "dc_key=" in flagsList:
+        retList.append(dc_key)
+    if "dc_pub=" in flagsList:
+        retList.append(dc_pub)
     return retList
 
 
@@ -556,12 +579,12 @@ def serverCmd(argv):
             directory, reqCert,
             expLabel, expLength, dhparam, psk, psk_ident, psk_hash, ssl3,
             max_ver, tickets, cipherlist, request_pha, require_pha, echo,
-            groups) = \
+            groups, dc_key, dc_pub) = \
         handleArgs(argv, "kctbvdlL",
                    ["reqcert", "param=", "psk=",
                     "psk-ident=", "psk-sha384", "ssl3", "max-ver=",
                     "tickets=", "cipherlist=", "request-pha", "require-pha",
-                    "echo", "groups="])
+                    "echo", "groups=", "dc_key=", "dc_pub="])
 
 
     if (cert_chain and not privateKey) or (not cert_chain and privateKey):
@@ -585,6 +608,8 @@ def serverCmd(argv):
         print("Using Tacks...")
     if reqCert:
         print("Asking for client certificates...")
+    if dc_key and dc_pub:
+        print("Usage of delegated credential is available...")
 
     #############
     sessionCache = SessionCache()
@@ -619,6 +644,8 @@ def serverCmd(argv):
         settings.dhGroups = dh_groups
         settings.eccCurves = ecc_groups
         settings.keyShares = []
+    if dc_key and dc_pub:
+        settings.delegated_credential = [SignatureScheme.ed25519]
 
     class MySimpleEchoHandler(BaseRequestHandler):
         def handle(self):
@@ -700,7 +727,9 @@ def handshake(self, connection):
                                               nextProtos=[b"http/1.1"],
                                               alpn=[bytearray(b'http/1.1')],
                                               reqCert=reqCert,
-                                              sni=sni)
+                                              sni=sni,
+                                              dc_key=dc_key,
+                                              dc_pub=dc_pub)
                                               # As an example (does not work here):
                                               #nextProtos=[b"spdy/3", b"spdy/2", b"http/1.1"])
                 try:
@@ -754,4 +783,3 @@ def handshake(self, connection):
         serverCmd(sys.argv[2:])
     else:
         printUsage("Unknown command: %s" % sys.argv[1])
-

tests/serverX509DCKey.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCT/Pf92sZZd5nP
+5WUs+ilhN2GxgSx1ylXyARar2MGWfYA/b0eMPsfVf+bLpQX/Da2drrq7L1HxTaId
+NUYsh/nvPLxROkA4AmECahFZSCkJZ6VdVgO2cMu7SNE9L7vZXIo22aL+COBqZ1xn
+vjMUyVXnrs02fMfcn6c/FIO0bd2YkRUiOP2iBAhCn6GmHZeJ0/OMHB7VJ0Ozpy63
+xsONxtIBgPEimQFlTCgmMmBjDzyquTIqVZ0ajQnHNKM4BaBcsu788sttPCh3i7EQ
+4sAI8Pw8tmPxs/nb2fMNgQEd+wjvIHTss82hzu9XUTDt9CAOWHyRCUi5D9a6dMS4
+oscRpQiPAgMBAAECggEAEy4gPiiSuJnFt6o1mMS7hDwXT1g8mO+mf/0gIRmwzX5q
+ls4nacfhQoyXLyGuS0ZMkDlLPmN9rVawgjSbab4d6KHojmaMWDYGuLdilD3EA9IJ
+HrW9OXIZFab0Z4e+Qwe5ai5+74na/C91TITcPf9yQNrpAfzeMnwGwyg3gbUTmWhR
+2AriFcTdWINmZKNKbOS0Fkw/Kn7zPAidjSanvthNdQPNi1k1fob0rj05Q57kEe4d
+lOC5j79mYV3XUWlyf6g8knmoBsQ1Y0gfFyro1R3VtAJSlU8KPJ41bhlE5kpiUZYb
+cWRCJYF3anRxTxo8getRLffE19en0i+z99tZIzgwoQKBgQDOgU5/2DK6ussAm9kv
+6IPOuwr9gEaYQAMmgG08ck2CCW7J/MGAGXURtZOMf5zSsMBvHXT/xvAWBuElilbF
+9/DNHz51Ts2GvyqPSzSFHxQUo3ljV6n/qIXP+7Xb3xwzjGVzXl2CfPeI0hxVZmz+
+W6jcvczqyOXxjH+WZFmgLnzJSQKBgQC3dTFQYSWJpXb9gVNSWLRbN6ZVgoEjMbOJ
+N8Pzj+3hcKYeQJUdH1dxbGdjFUrCrkHVWEaXZ7eH5B/fWd+uTBUgwnO0FExjpkAm
+EsakhP96vdN/jtax0S7S1Jcn6Ty6YdBAzXT9JsguxAUvVE4uSpdrdxpJF/4F6key
+WdoXE9JbFwKBgG52qvgmPVS3sPm9ZFuFRGSkl0dtg9XTgBvrXQOVnTJvO01fIF8W
+vxHfEHN6m/f0RqvplPlxgGI4Ad3j93DkpXIEQZPcuIJY5jpKn2iKbGJx4/ApJ62z
+hwjve6OG4H4OnwIsu1ae5IbS5gckyC7z9wtFmEULfD1Oy702Jt9RnrzJAoGAbtug
+SwQJHN4hwxpM8SutAJnmJzHPOycjaD2MaTeF9X6OwyUfdhOkUWPCLbuGC5IlMfg/
+3+nKm5EcOWkjoz1SXxNhu2Wwq16g0ODzrCK6Br+CeEgmMBlJhBj2piVoju/gWehN
+U1QGD0xgHbOB8rMcQNIdziFzXLuvS3TENsHBkU0CgYBg4mTavRdbMwDUN07aRpt8
++EaJJnmxWRl0wD9yIvFVZa0wPRYd54/HOMj5dqHHklgn1eb+HwxdyEYw/kUMn+oC
+roPHTA4hYXDMqAHkXMC6sDmG5MVgP6ozWEYVbiK/5y6QsTozLciNpwA0STGHmtcC
+kQpskxTgLWPCkzB6RvWqzw==
+-----END PRIVATE KEY-----

tests/serverX509DCPub.pem

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAk/z3/drGWXeZz+VlLPop
+YTdhsYEsdcpV8gEWq9jBln2AP29HjD7H1X/my6UF/w2tna66uy9R8U2iHTVGLIf5
+7zy8UTpAOAJhAmoRWUgpCWelXVYDtnDLu0jRPS+72VyKNtmi/gjgamdcZ74zFMlV
+567NNnzH3J+nPxSDtG3dmJEVIjj9ogQIQp+hph2XidPzjBwe1SdDs6cut8bDjcbS
+AYDxIpkBZUwoJjJgYw88qrkyKlWdGo0JxzSjOAWgXLLu/PLLbTwod4uxEOLACPD8
+PLZj8bP529nzDYEBHfsI7yB07LPNoc7vV1Ew7fQgDlh8kQlIuQ/WunTEuKLHEaUI
+jwIDAQAB
+-----END PUBLIC KEY-----

tests/tlstest.py

@@ -21,6 +21,8 @@
 import timeit
 import getopt
 from tempfile import mkstemp
+
+from tlslite.x509 import DelegatedCredential
 try:
     from BaseHTTPServer import HTTPServer
     from SimpleHTTPServer import SimpleHTTPRequestHandler
@@ -45,6 +47,7 @@
 import ssl
 from tlslite import *
 from tlslite.constants import KeyUpdateMessageType, ECPointFormat, SignatureScheme
+from tlslite.utils.pem import dePem
 
 try:
     from tack.structures.Tack import Tack
@@ -1910,6 +1913,21 @@ def heartbeat_response_check(message):
 
     test_no += 1
 
+    print("Test {0} - Delegated Credential test".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.maxVersion = (3, 4)
+    settings.dc_sig_algs = [SignatureScheme.rsa_pkcs1_sha256]
+    connection.handshakeClientCert(settings=settings)
+    assert connection.session.delegated_credential is not None
+    assert isinstance(connection.session.delegated_credential,
+                      DelegatedCredential)
+    testConnClient(connection)
+    connection.close()
+
+    test_no += 1
+
     print('Test {0} - good standard XMLRPC https client'.format(test_no))
     address = address[0], address[1]+1
     synchro.recv(1)
@@ -1967,6 +1985,8 @@ def heartbeat_response_check(message):
         print("Non-critical error: socket error trying to reach internet "
               "server: ", e)
 
+
+
     synchro.close()
 
     if not badFault:
@@ -2119,6 +2139,13 @@ def connect():
     with open(os.path.join(dir, "serverEd448Key.pem")) as f:
         x509Ed448Key = parsePEMKey(f.read(), private=True,
                                    implementations=["python"])
+    with open(os.path.join(dir, "serverX509DCKey.pem")) as f:
+        s = f.read()
+        x509DelegatedCredPrivKey = parsePEMKey(s, private=True,
+                                               implementations=["python"])
+    with open(os.path.join(dir, "serverX509DCPub.pem")) as f:
+        s = f.read()
+        x509DelegatedCredPubKey = dePem(s, "PUBLIC KEY")
 
     test_no = 0
 
@@ -3655,6 +3682,25 @@ def heartbeat_response_check(message):
 
     test_no +=1
 
+    print("Test {0} - Delegated Credential test".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.maxVersion = (3, 4)
+    settings.dc_sig_algs = [SignatureScheme.rsa_pkcs1_sha256]
+    connection.handshakeServer(certChain=x509Chain,
+                               privateKey=x509Key,
+                               dc_key=x509DelegatedCredPrivKey,
+                               dc_pub=x509DelegatedCredPubKey,
+                               settings=settings)
+    assert connection.session.delegated_credential is not None
+    assert isinstance(connection.session.delegated_credential,
+                      DelegatedCredential)
+    testConnServer(connection)
+    connection.close()
+
+    test_no += 1
+
     print("Tests {0}-{1} - XMLRPXC server".format(test_no, test_no + 2))
 
     address = address[0], address[1]+1

tlslite/constants.py

@@ -181,6 +181,7 @@ class ExtensionType(TLSEnum):
     post_handshake_auth = 49  # TLS 1.3
     signature_algorithms_cert = 50  # TLS 1.3
     key_share = 51  # TLS 1.3
+    delegated_credential = 34 # TLS 1.3, RFC 9345
     supports_npn = 13172
     tack = 0xF300
     renegotiation_info = 0xff01  # RFC 5746

tlslite/extensions.py

@@ -14,6 +14,7 @@
         PskKeyExchangeMode, CertificateType, GroupName, ECPointFormat, \
         HeartbeatMode, CertificateCompressionAlgorithm
 from .errors import TLSInternalError
+from .x509 import DelegatedCredential
 
 
 class TLSExtension(object):
@@ -1350,6 +1351,48 @@ def parse(self, p):
         return self
 
 
+class DelegatedCredentialCertExtension(TLSExtension):
+    """
+    Extension of CertificateEntry consisting delegated credential.
+
+    See RFC9345.
+
+    Creates, parses and returns the Delegated Credential class.
+    """
+    def __init__(self):
+        """Create instance of DelegatedCredentialCertExtension."""
+        super(DelegatedCredentialCertExtension, self).__init__(
+            extType=ExtensionType.delegated_credential)
+        self.delegated_credential = None
+
+    def create(self, delegated_credential):
+        """Set values of the extension."""
+        self.delegated_credential = delegated_credential
+        return self
+
+    def parse(self, p):
+        """Deserialise the data from on the wire representation."""
+        self.delegated_credential = DelegatedCredential().parse(p)
+        return self
+
+    @property
+    def extData(self):
+        """Serialise the object."""
+        writer = Writer()
+        writer.addFour(
+            self.delegated_credential.cred.valid_time)
+        writer.addOne(
+            self.delegated_credential.cred.dc_cert_verify_algorithm[0])
+        writer.addOne(
+            self.delegated_credential.cred.dc_cert_verify_algorithm[1])
+        writer.add_var_bytes(
+            self.delegated_credential.cred.subject_public_key_info, 3)
+        writer.addOne(self.delegated_credential.algorithm[0])
+        writer.addOne(self.delegated_credential.algorithm[1])
+        writer.add_var_bytes(self.delegated_credential.signature, 2)
+        return writer.bytes
+
+
 class SupportedGroupsExtension(VarListExtension):
     """
     Client side list of supported groups of (EC)DHE key exchage.
@@ -1370,7 +1413,7 @@ def __init__(self):
 
 class ECPointFormatsExtension(VarListExtension):
     """
-    Client side list of supported ECC point formats.
+    List of supported ECC point formats.
 
     See RFC4492.
 
@@ -1412,6 +1455,27 @@ def _list_to_repr(self):
         return "[{0}]".format(", ".join(values))
 
 
+class DelegatedCredentialExtension(_SigListExt):
+    """
+    Client side list of supported signature algorithms
+    for use of delegated credentials.
+    SignatureSchemeList.
+
+    See RFC8446 / RFC9345.
+
+    :vartype sigalgs: list of tuples
+    :ivar sigalgs: list of signature schemes supported by peer
+    """
+
+    def __init__(self):
+        """Create instance of class"""
+        super(DelegatedCredentialExtension, self).__init__(
+            1, 2, 2,
+            'sigalgs',
+            ExtensionType.delegated_credential,
+            SignatureScheme)
+
+
 class SignatureAlgorithmsExtension(_SigListExt):
     """
     Client side list of supported signature algorithms.
@@ -2190,7 +2254,8 @@ def __init__(self):
     ExtensionType.cookie: CookieExtension,
     ExtensionType.record_size_limit: RecordSizeLimitExtension,
     ExtensionType.session_ticket: SessionTicketExtension,
-    ExtensionType.compress_certificate: CompressedCertificateExtension
+    ExtensionType.compress_certificate: CompressedCertificateExtension,
+    ExtensionType.delegated_credential: DelegatedCredentialExtension
 }
 
 TLSExtension._serverExtensions = {
@@ -2202,7 +2267,9 @@ def __init__(self):
 }
 
 TLSExtension._certificateExtensions = {
-    ExtensionType.status_request: CertificateStatusExtension
+    ExtensionType.status_request: CertificateStatusExtension,
+    ExtensionType.delegated_credential: DelegatedCredentialCertExtension
+
 }
 
 TLSExtension._hrrExtensions = {