Add support for 2+ listeners certificates

Signed-off-by: Alex Zgabur <[email protected]>

pyproject.toml

@@ -19,7 +19,7 @@ PyJWT = "*"
 lxml = "*"
 cryptography = "*"
 backoff = "*"
-httpx = { version = "*", extras = ["http2"] }
+httpx = { version = "0.27.2", extras = ["http2"] }
 openshift-client = ">=2"
 apyproxy = "*"
 weakget = "*"

testsuite/gateway/__init__.py

@@ -135,7 +135,7 @@ def wait_for_ready(self, timeout: int = 90):
         """Waits until the gateway is ready"""
 
     @abstractmethod
-    def get_tls_cert(self) -> Optional[Certificate]:
+    def get_tls_cert(self, hostname: str) -> Optional[Certificate]:
         """Returns TLS cert bound to this Gateway, if the Gateway does not use TLS, returns None"""
 
 

testsuite/gateway/envoy/__init__.py

@@ -99,7 +99,7 @@ def commit(self):
         )
         self.service.commit()
 
-    def get_tls_cert(self) -> Optional[Certificate]:
+    def get_tls_cert(self, _) -> Optional[Certificate]:
         return None
 
     def delete(self):

testsuite/gateway/exposers.py

@@ -67,8 +67,9 @@ class LoadBalancerServiceExposer(Exposer):
     """Exposer using Load Balancer service for Gateway"""
 
     def expose_hostname(self, name, gateway: Gateway) -> Hostname:
+        hostname = f"{name}.{self.base_domain}"
         return StaticLocalHostname(
-            f"{name}.{self.base_domain}", gateway.external_ip, gateway.get_tls_cert(), force_https=self.passthrough
+            hostname, gateway.external_ip, gateway.get_tls_cert(hostname), force_https=self.passthrough
         )
 
     @property

testsuite/gateway/gateway_api/gateway.py

@@ -9,7 +9,7 @@
 from testsuite.kubernetes.client import KubernetesClient
 from testsuite.kubernetes import KubernetesObject, modify
 from testsuite.kuadrant.policy import Policy
-from testsuite.utils import check_condition, asdict
+from testsuite.utils import check_condition, asdict, domain_match
 
 
 class KuadrantGateway(KubernetesObject, Gateway):
@@ -84,11 +84,15 @@ def is_affected_by(self, policy: Policy) -> bool:
                 return True
         return False
 
-    def get_tls_cert(self):
-        if "tls" not in self.model.spec.listeners[0]:
+    def get_tls_cert(self, hostname):
+        tls_cert_secret_name = None
+        for listener in self.all_tls_listeners():
+            if domain_match(hostname, listener.hostname):
+                tls_cert_secret_name = listener.tls.certificateRefs[0].name
+
+        if tls_cert_secret_name is None:
             return None
 
-        tls_cert_secret_name = self.cert_secret_name
         try:
             tls_cert_secret = self.cluster.get_secret(tls_cert_secret_name)
         except oc.OpenShiftPythonException as e:
@@ -102,20 +106,24 @@ def get_tls_cert(self):
         )
         return tls_cert
 
+    def all_tls_listeners(self):
+        """Yields all listeners in gateway that support 'tls'"""
+        for listener in self.model.spec.listeners:
+            if "tls" in listener:
+                yield listener
+
     def delete(self, ignore_not_found=True, cmd_args=None):
         res = super().delete(ignore_not_found, cmd_args)
         with self.cluster.context:
             # TLSPolicy does not delete certificates it creates
-            oc.selector(f"secret/{self.cert_secret_name}").delete(ignore_not_found=True)
+            for secret in oc.selector("secret").objects():
+                if "tls" in secret.name() and self.name() in secret.name():
+                    secret.delete()
+
             # Istio does not delete ServiceAccount
             oc.selector(f"sa/{self.service_name}").delete(ignore_not_found=True)
         return res
 
-    @property
-    def cert_secret_name(self):
-        """Returns name of the secret with generated TLS certificate"""
-        return self.model.spec.listeners[0].tls.certificateRefs[0].name
-
     @property
     def reference(self):
         return {

testsuite/gateway/gateway_api/hostname.py

@@ -15,7 +15,7 @@
 class StaticHostname(Hostname):
     """Already exposed hostname object"""
 
-    def __init__(self, hostname, tls_cert_getter: Callable[[], Certificate | bool] = None):
+    def __init__(self, hostname, tls_cert_getter: Callable[[str], Certificate | bool] = None):
         """
         :param hostname: Hostname that is exposed
         :param tls_cert_getter: Function that will gather TLS certificate when called,
@@ -27,9 +27,11 @@ def __init__(self, hostname, tls_cert_getter: Callable[[], Certificate | bool] =
 
     def client(self, **kwargs) -> KuadrantClient:
         protocol = "http"
-        if self.tls_cert_getter is not None and self.tls_cert_getter() is not None:
-            protocol = "https"
-            kwargs.setdefault("verify", self.tls_cert_getter())
+        if self.tls_cert_getter is not None:
+            cert = self.tls_cert_getter(self.hostname)
+            if cert is not None:
+                protocol = "https"
+                kwargs.setdefault("verify", self.tls_cert_getter(self.hostname))
         return KuadrantClient(base_url=f"{protocol}://{self.hostname}", **kwargs)
 
     @property

testsuite/tests/singlecluster/gateway/test_external_ca.py

@@ -60,7 +60,7 @@ def cluster_issuer(testconfig, cluster):
 def client(hostname, gateway):
     """Returns httpx client to be used for requests, it also commits AuthConfig"""
     root_cert = resources.files("testsuite.resources").joinpath("letsencrypt-stg-root-x1.pem").read_text()
-    old_cert = gateway.get_tls_cert()
+    old_cert = gateway.get_tls_cert(hostname.hostname)
     cert = dataclasses.replace(old_cert, chain=old_cert.certificate + root_cert)
     client = hostname.client(verify=cert)
     yield client

testsuite/tests/singlecluster/gateway/tlspolicy/test_cert_parameters.py

@@ -29,9 +29,9 @@ def tls_policy(tls_policy):
 
 
 @pytest.fixture(scope="module")
-def tls_cert(gateway):  # pylint: disable=unused-argument
+def tls_cert(gateway, wildcard_domain):  # pylint: disable=unused-argument
     """Return certificate generated by TLSPolicy"""
-    return gateway.get_tls_cert()
+    return gateway.get_tls_cert(wildcard_domain)
 
 
 def test_tls_cert_common_name(tls_cert):

testsuite/utils.py

@@ -209,3 +209,20 @@ def sleep_ttl(hostname: str):
         return
 
     sleep(dns.resolver.resolve(hostname).rrset.ttl)  # type: ignore
+
+
+def domain_match(first: str, second: str):
+    """Returns true if domains are the same, considering left-most wildcard"""
+    # strip last '.'
+    if first[-1] == ".":
+        first = first[:-1]
+    if second[-1] == ".":
+        second = second[:-1]
+
+    if first == second:
+        return True
+    if first[0] == "*":
+        return first[2:] == ".".join(second.split(".")[1:])
+    if second[0] == "*":
+        return second[2:] == ".".join(first.split(".")[1:])
+    return False