outposts/controllers: k8s: sanitize resource names to comply with DNS subdomain standards (#13444)

* wip

* wip[skip ci]

* add some tests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

---------

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
Co-authored-by: Jens Langhammer <jens@goauthentik.io>

authentik/outposts/controllers/k8s/base.py

@@ -1,5 +1,6 @@
 """Base Kubernetes Reconciler"""
 
+import re
 from dataclasses import asdict
 from json import dumps
 from typing import TYPE_CHECKING, Generic, TypeVar
@@ -67,7 +68,8 @@ class KubernetesObjectReconciler(Generic[T]):
     @property
     def name(self) -> str:
         """Get the name of the object this reconciler manages"""
-        return (
+
+        base_name = (
             self.controller.outpost.config.object_naming_template
             % {
                 "name": slugify(self.controller.outpost.name),
@@ -75,6 +77,16 @@ class KubernetesObjectReconciler(Generic[T]):
             }
         ).lower()
 
+        formatted = slugify(base_name)
+        formatted = re.sub(r"[^a-z0-9-]", "-", formatted)
+        formatted = re.sub(r"-+", "-", formatted)
+        formatted = formatted[:63]
+
+        if not formatted:
+            formatted = f"outpost-{self.controller.outpost.uuid.hex}"[:63]
+
+        return formatted
+
     def get_patched_reference_object(self) -> T:
         """Get patched reference object"""
         reference = self.get_reference_object()
@@ -112,7 +124,6 @@ class KubernetesObjectReconciler(Generic[T]):
             try:
                 current = self.retrieve()
             except (OpenApiException, HTTPError) as exc:
-
                 if isinstance(exc, ApiException) and exc.status == HttpResponseNotFound.status_code:
                     self.logger.debug("Failed to get current, triggering recreate")
                     raise NeedsRecreate from exc
@@ -156,7 +167,6 @@ class KubernetesObjectReconciler(Generic[T]):
             self.delete(current)
             self.logger.debug("Removing")
         except (OpenApiException, HTTPError) as exc:
-
             if isinstance(exc, ApiException) and exc.status == HttpResponseNotFound.status_code:
                 self.logger.debug("Failed to get current, assuming non-existent")
                 return

authentik/outposts/controllers/kubernetes.py

@@ -61,9 +61,14 @@ class KubernetesController(BaseController):
     client: KubernetesClient
     connection: KubernetesServiceConnection
 
-    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection) -> None:
+    def __init__(
+        self,
+        outpost: Outpost,
+        connection: KubernetesServiceConnection,
+        client: KubernetesClient | None = None,
+    ) -> None:
         super().__init__(outpost, connection)
-        self.client = KubernetesClient(connection)
+        self.client = client if client else KubernetesClient(connection)
         self.reconcilers = {
             SecretReconciler.reconciler_name(): SecretReconciler,
             DeploymentReconciler.reconciler_name(): DeploymentReconciler,

authentik/outposts/tests/test_controller_k8s.py

@@ -0,0 +1,44 @@
+"""Kubernetes controller tests"""
+
+from django.test import TestCase
+
+from authentik.blueprints.tests import reconcile_app
+from authentik.lib.generators import generate_id
+from authentik.outposts.apps import MANAGED_OUTPOST
+from authentik.outposts.controllers.k8s.deployment import DeploymentReconciler
+from authentik.outposts.controllers.kubernetes import KubernetesController
+from authentik.outposts.models import KubernetesServiceConnection, Outpost, OutpostType
+
+
+class KubernetesControllerTests(TestCase):
+    """Kubernetes controller tests"""
+
+    @reconcile_app("authentik_outposts")
+    def setUp(self) -> None:
+        self.outpost = Outpost.objects.create(
+            name="test",
+            type=OutpostType.PROXY,
+        )
+        self.integration = KubernetesServiceConnection(name="test")
+
+    def test_gen_name(self):
+        """Ensure the generated name is valid"""
+        controller = KubernetesController(
+            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(),
+            self.integration,
+            # Pass something not-none as client so we don't
+            # attempt to connect to K8s as that's not needed
+            client=self,
+        )
+        rec = DeploymentReconciler(controller)
+        self.assertEqual(rec.name, "ak-outpost-authentik-embedded-outpost")
+
+        controller.outpost.name = generate_id()
+        self.assertLess(len(rec.name), 64)
+
+        # Test custom naming template
+        _cfg = controller.outpost.config
+        _cfg.object_naming_template = ""
+        controller.outpost.config = _cfg
+        self.assertEqual(rec.name, f"outpost-{controller.outpost.uuid.hex}")
+        self.assertLess(len(rec.name), 64)