dt: Allow selecting key type for certificates in DT tests

This change permits users of the TLSCertManager to select which type of
key to use in certificates.  If one isn't selected, a random one is
chosen, in the hope this increases our test coverage over time.

Additionally, the tls_version_test has been updated to test both ECDSA
and RSA based certificates.

Signed-off-by: Michael Boquard <michael@redpanda.com>

tests/rptest/services/tls.py

@@ -7,6 +7,8 @@
 import string
 import random
 
+from enum import Enum
+
 _ca_config_tmpl = """
 # OpenSSL CA configuration file
 [ ca ]
@@ -302,6 +304,11 @@
     "Certificate", ["cfg", "key", "crt", "ca", "p12_file", "p12_password"])
 
 
+class TLSKeyType(Enum):
+    RSA = 0,
+    ECDSA = 1
+
+
 class TLSCertManager:
     """
     When a TLSCertManager is instantiated a new CA is automatically created and
@@ -312,11 +319,20 @@ class TLSCertManager:
     it is common for clients to take paths to these files, it is best to keep
     the instance alive for as long as the files are in use.
     """
-    def __init__(self, logger, ca_end_date=None, cert_expiry_days=1):
+    def __init__(self,
+                 logger,
+                 ca_end_date=None,
+                 cert_expiry_days=1,
+                 key_type: typing.Optional[TLSKeyType] = None):
         self._logger = logger
         self._dir = tempfile.TemporaryDirectory()
         self._ca_end_date = ca_end_date
         self._cert_expiry_days = cert_expiry_days
+        # Before this change, all ducktape tests would use RSA based certificates
+        # In order to attempt to cover more code paths, we will randomly select
+        # between RSA and ECDSA if the key type is not specified
+        self._key_type: TLSKeyType = key_type or random.choice(
+            list(TLSKeyType))
         self._ca = self._create_ca()
         self.certs: dict[str, Certificate] = {}
 
@@ -354,6 +370,15 @@ def _exec(self, cmd):
 
             retries += 1
 
+    def _generate_key(self, out: str) -> None:
+        if self._key_type == TLSKeyType.RSA:
+            self._exec(f"openssl genrsa -out {out} 2048")
+        elif self._key_type == TLSKeyType.ECDSA:
+            self._exec(
+                f"openssl ecparam -name prime256v1 -genkey -noout -out {out}")
+        else:
+            raise ValueError(f"Unknown key type: {self._key_type}")
+
     def _create_ca(self) -> CertificateAuthority:
         cfg = self._with_dir("ca.conf")
         key = self._with_dir("ca.key")
@@ -365,7 +390,7 @@ def _create_ca(self) -> CertificateAuthority:
         with open(f"{cfg}", "w") as f:
             f.write(_ca_config_tmpl.format(dir=self._dir.name))
 
-        self._exec(f"openssl genrsa -out {key} 2048")
+        self._generate_key(key)
         self._exec(f"openssl req -new -x509 -config {cfg} "
                    f"-key {key} -out {crt} -days 365 -batch")
 
@@ -418,7 +443,7 @@ def create_cert(self,
             f.write(
                 _node_config_tmpl.format(host=host, common_name=common_name))
 
-        self._exec(f"openssl genrsa -out {key} 2048")
+        self._generate_key(key)
 
         self._exec(f"openssl req -new -config {cfg} "
                    f"-key {key} -out {csr} -batch")
@@ -482,12 +507,18 @@ def __init__(self,
                  logger,
                  chain_len=2,
                  cert_expiry_days=1,
-                 ca_expiry_days=7):
+                 ca_expiry_days=7,
+                 key_type: typing.Optional[TLSKeyType] = None):
         assert chain_len > 0
         self._logger = logger
         self._dir = tempfile.TemporaryDirectory()
         self.cert_expiry_days = cert_expiry_days
         self.ca_expiry_days = ca_expiry_days
+        # Before this change, all ducktape tests would use RSA based certificates
+        # In order to attempt to cover more code paths, we will randomly select
+        # between RSA and ECDSA if the key type is not specified
+        self._key_type: TLSKeyType = key_type or random.choice(
+            list(TLSKeyType))
         self._cas: list[CertificateAuthority] = []
         self._cas.append(
             self._create_ca(
@@ -550,8 +581,10 @@ def _create_ca(self,
         with open(srl, 'w') as f:
             f.writelines(["01"])
 
+        self._generate_key(key)
+
         self._exec(
-            f"openssl req -new -nodes -config {cfg} -out {csr} -keyout {key}")
+            f"openssl req -new -nodes -config {cfg} -out {csr} -key {key}")
         self._exec(
             f"openssl ca {'-selfsign' if selfsign else ''} -config {parent_cfg} "
             f"-in {csr} -out {crt} -extensions {ext} -days {days} -batch")
@@ -609,8 +642,9 @@ def create_cert(self,
             f.write(
                 _server_config_tmpl.format(host=host, common_name=common_name))
 
+        self._generate_key(key)
         self._exec(f"openssl req -new -nodes -config {cfg} "
-                   f"-keyout {key} -out {csr} -batch")
+                   f"-key {key} -out {csr} -batch")
         self._exec(
             f"faketime -f {faketime} openssl ca -config {self.signing_ca.cfg} -policy match_pol "
             f"-in {csr} -out {crt} -extensions server_ext -days {self.cert_expiry_days} -batch"

tests/rptest/tests/tls_version_test.py

@@ -20,7 +20,7 @@
 from rptest.services.cluster import cluster
 from rptest.services.redpanda import TLSProvider, SecurityConfig
 from rptest.services.redpanda_installer import RedpandaInstaller
-from rptest.services.tls import Certificate, CertificateAuthority, TLSCertManager
+from rptest.services.tls import Certificate, CertificateAuthority, TLSCertManager, TLSKeyType
 from rptest.tests.redpanda_test import RedpandaTest
 
 
@@ -83,10 +83,10 @@ class TLSVersionTestBase(RedpandaTest):
     """
     Base test class that sets up TLS on the Kafka API interface
     """
-    def __init__(self, test_context):
+    def __init__(self, test_context, key_type: TLSKeyType):
         super(TLSVersionTestBase, self).__init__(test_context=test_context)
         self.security = SecurityConfig()
-        self.tls = TLSCertManager(self.logger)
+        self.tls = TLSCertManager(self.logger, key_type=key_type)
         self.admin = Admin(self.redpanda)
         self.installer = self.redpanda._installer
 
@@ -122,8 +122,6 @@ def verify_tls_version(self, node: ClusterNode, tls_version: TLSVersion,
                 assert self._output_error(e.output.decode(
                 )), f"Output not expected for failure: {e.output.decode()}"
 
-
-class TLSVersionTest(TLSVersionTestBase):
     @cluster(num_nodes=3, log_allow_list=PERMITTED_ERROR_MESSAGE)
     @matrix(version=[0, 1, 2, 3])
     def test_change_version(self, version: int):
@@ -150,3 +148,15 @@ def test_change_version(self, version: int):
                 self.verify_tls_version(node=n,
                                         tls_version=v,
                                         expect_fail=expect_failure)
+
+
+class TLSVersionTestRSA(TLSVersionTestBase):
+    def __init__(self, test_context):
+        super(TLSVersionTestRSA, self).__init__(test_context=test_context,
+                                                key_type=TLSKeyType.RSA)
+
+
+class TLSVersionTestECDSA(TLSVersionTestBase):
+    def __init__(self, test_context):
+        super(TLSVersionTestECDSA, self).__init__(test_context=test_context,
+                                                  key_type=TLSKeyType.ECDSA)