feat: Support certificateAuthorityCertSecretRef on TwingateResource CRD

app/conftest.py

@@ -1,8 +1,11 @@
 from unittest.mock import MagicMock, patch
 
+import kubernetes
 import pytest
 import responses
 
+from app.api.tests.factories import BASE64_OF_VALID_CA_CERT
+
 
 @pytest.fixture
 def mocked_responses():
@@ -38,3 +41,12 @@ def k8s_apps_client_mock():
     with patch("kubernetes.client.AppsV1Api") as k8sclient_mock:
         k8sclient_mock.return_value = client_mock
         yield client_mock
+
+
+@pytest.fixture
+def k8s_secret_mock():
+    return kubernetes.client.V1Secret(
+        type="kubernetes.io/tls",
+        metadata=kubernetes.client.V1ObjectMeta(name="gateway-tls"),
+        data={"ca.crt": BASE64_OF_VALID_CA_CERT},
+    )

app/crds.py

@@ -1,9 +1,11 @@
+import base64
 import logging
 from collections.abc import MutableMapping
 from datetime import datetime
 from enum import Enum, StrEnum
 from typing import Annotated, Any, cast
 
+import kopf
 import kubernetes.client
 import pendulum
 from croniter import croniter
@@ -21,6 +23,8 @@
 from semantic_version import NpmSpec
 
 from app.settings import get_settings
+from app.utils import validate_pem_x509_certificate
+from app.utils_k8s import k8s_read_namespaced_secret
 from app.version_policy_providers import get_provider
 
 K8sObject = MutableMapping[Any, Any]
@@ -122,8 +126,39 @@ class ResourceProxy(BaseModel):
 
     address: str
     certificate_authority_cert: Annotated[
-        Base64Str, AfterValidator(lambda v: v.strip())
-    ]
+        Base64Str | None, AfterValidator(lambda v: v.strip() if v is not None else None)
+    ] = None
+    certificate_authority_cert_secret_ref: _KubernetesObjectRef | None = None
+
+    def get_certificate_authority_cert(self) -> str | None:
+        if secret_ref := self.certificate_authority_cert_secret_ref:
+            if secret := k8s_read_namespaced_secret(
+                secret_ref.namespace, secret_ref.name
+            ):
+                return self.read_certificate_authority_cert_from_secret(secret)
+
+            return None
+
+        return self.certificate_authority_cert
+
+    @staticmethod
+    def read_certificate_authority_cert_from_secret(
+        secret: kubernetes.client.V1Secret,
+    ) -> str:
+        secret_name = secret.metadata.name
+        if not (ca_cert := secret.data.get("ca.crt")):
+            raise kopf.PermanentError(
+                f"Kubernetes Secret object: {secret_name} is missing ca.crt."
+            )
+
+        try:
+            ca_cert = base64.b64decode(ca_cert).decode()
+            validate_pem_x509_certificate(ca_cert)
+            return ca_cert
+        except ValueError as ex:
+            raise kopf.PermanentError(
+                f"Kubernetes Secret object: {secret_name} ca.crt is invalid."
+            ) from ex
 
 
 class ResourceType(StrEnum):
@@ -190,9 +225,14 @@ def to_graphql_arguments(
                 }
             case ResourceType.KUBERNETES:
                 resource_proxy = cast(ResourceProxy, self.proxy)
+                ca_cert = resource_proxy.get_certificate_authority_cert()
+                if ca_cert is None:
+                    raise kopf.PermanentError(
+                        "Certificate authority cert is not found for Kubernetes Resource type"
+                    )
                 graphql_args |= {
                     "proxy_address": resource_proxy.address,
-                    "certificate_authority_cert": resource_proxy.certificate_authority_cert,
+                    "certificate_authority_cert": ca_cert,
                 }
 
         return graphql_args

app/handlers/handlers_services.py

@@ -7,8 +7,9 @@
 import kubernetes
 from kopf import Body, Status
 
-from app.crds import ResourceType
-from app.utils import to_bool, validate_pem_x509_certificate
+from app.crds import ResourceProxy, ResourceType
+from app.utils import to_bool
+from app.utils_k8s import k8s_read_namespaced_secret
 
 
 def k8s_get_twingate_resource(
@@ -25,40 +26,6 @@ def k8s_get_twingate_resource(
         raise
 
 
-def k8s_get_tls_secret(namespace: str, name: str) -> kubernetes.client.V1Secret | None:
-    try:
-        return kubernetes.client.CoreV1Api().read_namespaced_secret(
-            name=name, namespace=namespace
-        )
-    except kubernetes.client.exceptions.ApiException as ex:
-        if ex.status == 404:
-            return None
-
-        raise
-
-
-def get_ca_cert(tls_secret: kubernetes.client.V1Secret) -> str:
-    tls_secret_name = tls_secret.metadata.name
-    if tls_secret.type != "kubernetes.io/tls":
-        raise kopf.PermanentError(
-            f"Kubernetes Secret object: {tls_secret_name} type is invalid."
-        )
-
-    if not (ca_cert := tls_secret.data.get("ca.crt")):
-        raise kopf.PermanentError(
-            f"Kubernetes Secret object: {tls_secret_name} is missing ca.crt."
-        )
-
-    try:
-        validate_pem_x509_certificate(base64.b64decode(ca_cert).decode())
-    except ValueError as ex:
-        raise kopf.PermanentError(
-            f"Kubernetes Secret object: {tls_secret_name} ca.crt is invalid."
-        ) from ex
-
-    return ca_cert
-
-
 ALLOWED_EXTRA_ANNOTATIONS: list[tuple[str, Callable]] = [
     ("name", str),
     ("alias", str),
@@ -122,14 +89,14 @@ def service_to_twingate_resource(service_body: Body, namespace: str) -> dict:
             result["spec"][key] = convert_f(value)
 
     if result["spec"].get("type") == ResourceType.KUBERNETES:
-        if not (tls_secret_name := meta.annotations.get(TLS_OBJECT_ANNOTATION)):
+        if not (secret_name := meta.annotations.get(TLS_OBJECT_ANNOTATION)):
             raise kopf.PermanentError(
                 f"{TLS_OBJECT_ANNOTATION} annotation is not provided."
             )
 
-        if not (tls_secret := k8s_get_tls_secret(namespace, tls_secret_name)):
+        if not (secret := k8s_read_namespaced_secret(namespace, secret_name)):
             raise kopf.PermanentError(
-                f"Kubernetes Secret object: {tls_secret_name} is missing."
+                f"Kubernetes Secret object: {secret_name} is missing."
             )
 
         result["spec"] |= {
@@ -140,7 +107,11 @@ def service_to_twingate_resource(service_body: Body, namespace: str) -> dict:
                     if spec["type"] == ServiceType.LOAD_BALANCER
                     else f"{service_name}.{namespace}.svc.cluster.local"
                 ),
-                "certificateAuthorityCert": get_ca_cert(tls_secret),
+                "certificateAuthorityCert": base64.b64encode(
+                    ResourceProxy.read_certificate_authority_cert_from_secret(
+                        secret
+                    ).encode()
+                ).decode(),
             },
         }
 

app/handlers/tests/test_handlers_services.py

@@ -7,11 +7,10 @@
 from kopf._core.intents.causes import Reason
 
 from app.api.tests.factories import BASE64_OF_VALID_CA_CERT
-from app.crds import ResourceType
+from app.crds import ResourceProxy, ResourceType
 from app.handlers.handlers_services import (
     ALLOWED_EXTRA_ANNOTATIONS,
     TLS_OBJECT_ANNOTATION,
-    k8s_get_tls_secret,
     k8s_get_twingate_resource,
     service_to_twingate_resource,
     twingate_service_create,
@@ -114,15 +113,6 @@ def example_load_balancer_gateway_service_body():
     return kopf.Body(yaml.safe_load(yaml_str))
 
 
-@pytest.fixture
-def k8s_tls_secret_mock():
-    return kubernetes.client.V1Secret(
-        type="kubernetes.io/tls",
-        metadata=kubernetes.client.V1ObjectMeta(name="gateway-tls"),
-        data={"ca.crt": BASE64_OF_VALID_CA_CERT},
-    )
-
-
 @pytest.fixture
 def k8s_customobjects_client_mock():
     client_mock = MagicMock()
@@ -185,15 +175,21 @@ def test_kubernetes_resource_type_annotation(
         self,
         example_cluster_ip_gateway_service_body,
         k8s_core_client_mock,
-        k8s_tls_secret_mock,
+        k8s_secret_mock,
     ):
         tls_object_name = "gateway-tls"
         namespace = "custom-namespace"
-        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_tls_secret_mock
+        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_secret_mock
 
-        result = service_to_twingate_resource(
-            example_cluster_ip_gateway_service_body, namespace
-        )
+        with patch(
+            "app.handlers.handlers_services.ResourceProxy.read_certificate_authority_cert_from_secret",
+            wraps=ResourceProxy.read_certificate_authority_cert_from_secret,
+        ) as read_ca_cert_mock:
+            result = service_to_twingate_resource(
+                example_cluster_ip_gateway_service_body, namespace
+            )
+
+        read_ca_cert_mock.assert_called_once_with(k8s_secret_mock)
         k8s_core_client_mock.read_namespaced_secret.assert_called_once_with(
             namespace=namespace, name=tls_object_name
         )
@@ -248,59 +244,6 @@ def test_kubernetes_resource_type_annotation_without_k8s_secret_object(
                 example_cluster_ip_gateway_service_body, "default"
             )
 
-    def test_kubernetes_resource_type_annotation_with_invalid_secret_type(
-        self,
-        example_cluster_ip_gateway_service_body,
-        k8s_core_client_mock,
-        k8s_tls_secret_mock,
-    ):
-        k8s_tls_secret_mock.type = "kubernetes.io/token"
-        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_tls_secret_mock
-
-        with pytest.raises(
-            kopf.PermanentError,
-            match=r"Kubernetes Secret object: gateway-tls type is invalid.",
-        ):
-            service_to_twingate_resource(
-                example_cluster_ip_gateway_service_body, "default"
-            )
-
-    def test_kubernetes_resource_type_annotation_without_ca_cert(
-        self,
-        example_cluster_ip_gateway_service_body,
-        k8s_core_client_mock,
-        k8s_tls_secret_mock,
-    ):
-        k8s_tls_secret_mock.data = {}
-        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_tls_secret_mock
-
-        with pytest.raises(
-            kopf.PermanentError,
-            match=r"Kubernetes Secret object: gateway-tls is missing ca.crt.",
-        ):
-            service_to_twingate_resource(
-                example_cluster_ip_gateway_service_body, "default"
-            )
-
-    def test_kubernetes_resource_type_annotation_with_invalid_ca_cert(
-        self,
-        example_cluster_ip_gateway_service_body,
-        k8s_core_client_mock,
-        k8s_tls_secret_mock,
-    ):
-        k8s_tls_secret_mock.data["ca.crt"] = (
-            "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tIE1JSUZmakNDQTJhZ0F3SUJBZ0lVQk50IC0tLS0tRU5EIENFUlRJRklDQVRFLS0tLS0="
-        )
-        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_tls_secret_mock
-
-        with pytest.raises(
-            kopf.PermanentError,
-            match=r"Kubernetes Secret object: gateway-tls ca.crt is invalid.",
-        ):
-            service_to_twingate_resource(
-                example_cluster_ip_gateway_service_body, "default"
-            )
-
     @pytest.mark.parametrize(
         ("status", "expected"),
         [
@@ -315,13 +258,13 @@ def test_kubernetes_resource_with_load_balancer_service_type(
         self,
         example_load_balancer_gateway_service_body,
         k8s_core_client_mock,
-        k8s_tls_secret_mock,
+        k8s_secret_mock,
         status,
         expected,
     ):
         tls_object_name = "gateway-tls"
         namespace = "default"
-        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_tls_secret_mock
+        k8s_core_client_mock.read_namespaced_secret.return_value = k8s_secret_mock
 
         with patch(
             "kopf._cogs.structs.bodies.Body.status",
@@ -406,21 +349,6 @@ def test_reraises_non_404_exceptions(self, k8s_customobjects_client_mock):
             k8s_get_twingate_resource("default", "test")
 
 
-class TestK8sGetTLSSecret:
-    def test_handles_404_returns_none(self, k8s_core_client_mock):
-        k8s_core_client_mock.read_namespaced_secret.side_effect = (
-            kubernetes.client.exceptions.ApiException(status=404)
-        )
-        assert k8s_get_tls_secret("default", "test") is None
-
-    def test_reraises_non_404_exceptions(self, k8s_core_client_mock):
-        k8s_core_client_mock.read_namespaced_secret.side_effect = (
-            kubernetes.client.exceptions.ApiException(status=500)
-        )
-        with pytest.raises(kubernetes.client.exceptions.ApiException):
-            k8s_get_tls_secret("default", "test")
-
-
 class TestTwingateServiceCreate:
     def test_create_service_triggers_creation_of_twingate_resource(
         self, example_service_body, kopf_handler_runner, k8s_customobjects_client_mock